Async (Asynq)¶

All background work runs through one system — Asynq, a Redis-backed task queue. The same Asynq deployment runs the server (consumes event-driven and cron-enqueued tasks) and the scheduler (enqueues cron-driven tasks on a wall-clock schedule).

Source: backend/internal/async/. Layout:

internal/async/
├── tasks_event.go      ← event task constants + payloads (location enrich, engine signals)
├── tasks_item.go       ← plan-item task constants (item:parse, link:enrich)
├── tasks_cron.go       ← cron task constants
├── tasks_cleanup.go    ← cleanup task constants + payloads (S3 delete, orphan sweep)
├── client.go           ← Asynq client (event-driven enqueue)
├── server.go           ← Asynq server + Handler interface
├── scheduler.go        ← Asynq scheduler + cronEntries
├── app.go              ← lifecycle wrapper (Start / Shutdown)
└── handlers/
    ├── handlers.go     ← HandlerService struct + constructor + Register
    ├── event.go        ← event handler interfaces (LocationEnricher, EngineEvaluator,
    │                     EngineExpirer, EngineReplayer) + methods + registerEventHandlers
    ├── item.go         ← plan-item handler methods
    ├── cleanup.go      ← S3 delete + orphan-sweep handler methods
    └── cron.go         ← cron handler methods + registerCronHandlers

internal/async/ stays free of services imports so service callers (which already use services) can enqueue without an import cycle. The handlers/ sub-package owns the services-aware code.

The event/cron file split is a consistent convention — anything event-driven lives in *_event.go / event.go; anything cron-driven lives in *_cron.go / cron.go. Same HandlerService type in both handler files (Go methods can live across files in the same package).

Server¶

// async/server.go
asynq.NewServer(redisOpt, asynq.Config{
    Concurrency: cfg.Worker.Concurrency,
    Queues: map[string]int{
        "critical": 6,
        "default":  3,
        "low":      1,
    },
    ErrorHandler: ...,  // logs failed tasks
})

Weights are relative dequeue ratios — Asynq pulls 6:3:1 from critical:default:low. Concurrency is how many tasks the server processes simultaneously across all queues. Per-task duration + outcome is recorded against the central Prometheus registry via observability.AsyncMetricsMiddleware.

Retries, exponential backoff, and the archived (dead-letter) queue are handled by Asynq itself. Failed tasks land in asynq:{<queue>}:archived with the captured error.

Scheduler¶

// async/scheduler.go
sched.Register("@every 5m", asynq.NewTask(TaskStatusUpdate, nil),
    asynq.MaxRetry(3),
    asynq.Unique(4*time.Minute),
    asynq.Retention(24*time.Hour),
)

asynq.Unique(window) is the cross-replica dedup primitive — every replica's scheduler tries to enqueue at the cron tick; the first to acquire Asynq's unique-task lock wins, the rest get ErrDuplicateTask and silently skip. The window is set shorter than the cron interval so the lock expires before the next tick.

Cron entries¶

Registered in async/scheduler.go::cronEntries:

Task	Interval	What it does
`cron:message_expiry`	30s	Hard-deletes disappearing chat messages past their `expires_at`
`cron:status_update`	5m	Recomputes event status (upcoming → live → finished)
`cron:redis_sync`	5m	Re-syncs event geo locations into Redis from Postgres
`cron:moment_cleanup`	5m	Soft-deletes expired non-journaled moments
`cron:token_cleanup`	1h	Deletes expired refresh tokens, OTPs, and sessions
`cron:account_suspension`	12h	Suspends accounts flagged for moderation
`cron:purge`	24h	Cleans orphaned events + purges archived events older than 14 days
`cron:deactivated_acct_purge`	24h	Hard-deletes accounts after the deactivation grace period
`cron:moment_purge`	24h	Hard-deletes soft-deleted moments older than 7 days
`cron:challenge_archive`	24h	Flips status=published → archived for time-bounded challenges past their window
`cron:user_stats_refresh`	1m	Recomputes `tomoda_users_*` Prometheus gauges (total / active_24h / new_24h / by_country) for the analytics dashboard
`cron:notification_purge`	24h	Deletes old read / expired notifications
`cron:parsed_link_sweep`	24h	GCs shared `parsed_links` rows no item references (item hard-deletes cascade `item_links` at the DB level, bypassing the inline cleanup)
`cron:location_enrich_sweep`	1m	Enqueues up to 2000 never-enriched locations (`enrichment_at IS NULL`) for the async equalizer. The self-heal net: seed rows leave `enrichment_at` nil and are drained in rate-limited waves; also backstops any runtime fire that failed. Asynq dedups per location id
`cron:location_embedded_count`	5m	Samples the count of active rows carrying a semantic embedding into the `tomoda_locations_embedded_total` gauge. A cheap COUNT; the graduation signal for the two-stage vector recall (`SearchLocationsSemantic`). As the embedded set grows, the coarse binary quantization behind stage 1 starts dropping true neighbors at a fixed rerank pool size, so watch this gauge to know when the binary-HNSW index should move to a disk-backed quantized index
`cron:ranking_label_prune`	24h	Deletes ranking impressions + selections older than 180 days, keeping the durable search-ranking label sink bounded

Adding a cron task means appending one entry to cronEntries AND registering a handler in async/handlers/handlers.go::Register.

Event-driven tasks¶

Task	Enqueued by	Notes
`item:parse`	`plan.ItemService` via `async.Client.EnqueueItemParse`	Default queue, per-item dedup ID, 3-minute task timeout. Deterministic capture parse; links fan out concurrently inside the handler.
`link:enrich`	`plan.ItemService` via `async.Client.EnqueueLinkEnrich`	Low queue, per-item dedup ID (force reparse undeduped), 3-minute task timeout. Additionally guarded by the `enrich:item:lock:*` Redis lock so a double fire can't double-spend the enrichment quota.
`location:enrich`	`location.LocationService` via `async.Client.EnqueueEnrichment`	Low queue, per-location dedup ID — back-to-back enqueues for the same location collapse into one execution
`cleanup:s3_delete`	`assets.Service.EnqueueDelete` (`backend/internal/platform/assets`)	Hard-delete cascades fan out one task per asset key. Low queue. The handler refuses curated-prefix keys (`stamps/`, `stamps/bespoke/`, `curios/`) as defence-in-depth.
`cleanup:s3_orphan_sweep`	Admin via `POST /api/v1/admin/async/s3-orphan-sweep?prefix=…`	One prefix per task; enumeration is bounded. Not scheduled — see orphan sweep rationale.
`engine:eval_travel_log`, `engine:eval_user_activated`	`gameengine.Signal` on travel-log write / user activation	Evaluate the challenge + reward rule trees and grant earned curios. See Services → Game Engine.
`engine:expire_challenge`, `engine:replay_challenge`, `engine:replay_stamp`	`gameengine` expiry scheduler / admin replay	Async challenge cleanup and rule replay.

Cleanup tasks¶

Both cleanup task types live in async/tasks_cleanup.go; handlers in async/handlers/cleanup.go.

type S3DeletePayload struct {
    Key       string    `json:"key"`
    Triggered time.Time `json:"triggered"`
}

type S3OrphanSweepPayload struct {
    Prefix    string    `json:"prefix,omitempty"`
    Triggered time.Time `json:"triggered"`
}

Task Handler invariants

cleanup:s3_delete Rejects empty keys and curated-prefix keys with asynq.SkipRetry. S3 errors propagate so asynq retries per MaxRetry, then archives on exhaustion.

cleanup:s3_orphan_sweep Looks up the prefix's owning column in orphanColumns() (no entry for curated prefixes), SELECT DISTINCT col FROM table ... Unscoped() to build the "owned" set (soft-deleted rows still own their assets), then ListObjectsV2 paginated against the prefix. Each unowned key gets enqueued as a cleanup:s3_delete so the sweep stays read-only and per-key retries apply uniformly. Skips with nil (not retry) when the bound FileStorage isn't an *S3Storage — local dev / MinIO.

The orphan sweep's per-prefix DB lookup map is the source of truth for "which DB column owns this prefix's keys" — adding a user-generated prefix means adding both a storage.Prefix constant and an entry to orphanColumns().

Lifecycle¶

async.App owns the runtime. App.Start() registers handlers + cron entries then runs Server and Scheduler in goroutines. App.Shutdown() stops them in reverse order. Both are idempotent.

The single binary serves multiple roles via SERVER_MODE; async.App.Start is only invoked when modes.runAsync is true (modes full and async). See cmd/server/main.go.

Registering a new task¶

Pick the path that matches your trigger.

Event-driven (request handler / service enqueues)¶

Define the task type and payload in async/tasks_event.go:

const TaskMyEvent = "my_domain:event"

type MyEventPayload struct {
    SomeID uuid.UUID `json:"some_id"`
}

Write the handler in async/handlers/event.go (methods on the existing HandlerService):

func (h *HandlerService) HandleMyEvent(ctx context.Context, t *asynq.Task) error {
    var payload async.MyEventPayload
    if err := json.Unmarshal(t.Payload(), &payload); err != nil {
        return fmt.Errorf("unmarshal: %v: %w", err, asynq.SkipRetry)
    }
    return h.someService.DoThing(ctx, payload.SomeID)
}

If the handler needs a service that isn't on HandlerService yet, declare a narrow interface in event.go (see LocationEnricher / EngineEvaluator for the pattern) and bind the concrete domain type to it in wiring/providers.go (AsyncSet or AppSet).

Wire it on the mux by adding one line to registerEventHandlers in event.go:

mux.HandleFunc(async.TaskMyEvent, h.HandleMyEvent)

Enqueue from a service via the injected *async.Client:

asyncClient.Enqueue(asynq.NewTask(async.TaskMyEvent, body), asynq.Queue("default"))

Cron-driven (scheduler enqueues on a wall-clock tick)¶

Define the task type in async/tasks_cron.go (cron tasks typically have no payload):

const TaskMyCron = "cron:my_cron"

Write the handler in async/handlers/cron.go:

func (h *HandlerService) HandleMyCron(ctx context.Context, _ *asynq.Task) error {
    return h.someService.DoSweep(ctx)
}

Wire it on the mux by adding one line to registerCronHandlers in cron.go:

mux.HandleFunc(async.TaskMyCron, h.HandleMyCron)

Append a cronEntry to cronEntries in async/scheduler.go. Use a cron spec (@every 1h, 0 3 * * *) and a uniqueWindow strictly shorter than the interval so the cross-replica dedup lock expires before the next tick:

{"@every 1h", asynq.NewTask(TaskMyCron, nil), 55 * time.Minute},

Common notes¶

Return asynq.SkipRetry from a handler to send the task straight to the archived queue — useful for un-recoverable errors like unmarshal failures or invalid IDs.

Operations¶

Recovery — failed tasks land in Asynq's archived queue per queue. Inspect / requeue via the admin DLQ endpoints below, or via asynqctl / the asynqmon web UI (neither deployed today, both compatible with our data structures).
Scaling — tomoda-async deployment runs N replicas. All replicas consume from the shared Asynq queues; the scheduler's asynq.Unique dedup ensures cron tasks only fire once per tick regardless of replica count.

Admin DLQ endpoints¶

admin.AsyncHandler (backend/internal/services/admin/async_handler.go) wraps the asynq Inspector + Client so an on-call admin can triage archived tasks without shelling into the cluster.

Method	Path	What it does
`GET`	`/api/v1/admin/async/queues`	Snapshot of every queue (size / active / pending / scheduled / retry / archived / completed / latency / paused)
`GET`	`/api/v1/admin/async/queues/:queue/archived`	Paginated archived listing — `page`, `page_size` (max 200); shows payload + last error
`POST`	`/api/v1/admin/async/queues/:queue/archived/:id/run`	Pull a single archived task back into pending
`POST`	`/api/v1/admin/async/queues/:queue/archived/run-all`	Bulk replay every archived task in a queue
`DELETE`	`/api/v1/admin/async/queues/:queue/archived/:id`	Drop a poison-payload task
`POST`	`/api/v1/admin/async/s3-orphan-sweep?prefix=…`	Enqueue an on-demand orphan sweep for one prefix (curated prefixes return 400)

Endpoints are generic across task types — the cleanup queue is not the only DLQ in play. All routes are mounted under the admin access gate (access.Require(access.TomodaAdmin)) in backend/internal/wiring/router.go.