Task queue on Postgres

Deployed There is no separate queue broker — no Redis Streams job queue, no RabbitMQ, no Celery. The table generation.tasks is the queue, and Postgres row-locking (SELECT … FOR UPDATE SKIP LOCKED) gives it exactly-one-claim semantics. A generation is enqueued as a row, a tool worker claims it, works it, and marks it terminal — all through the core-database data plane, which owns the only Postgres credentials in the fleet.

The whole contract lives in one proto: generation/task/v1/task.proto declares the GenerationDataService RPCs (each a SQL-annotated DB-codegen handler), and its sibling schema.sql holds the DDL — the table, the status enum, and the claim index.

The table is the queue

A task moves through a small Postgres enum, generation.task_status:

queued → processing → completed | failed

dispatched is also a legal enum value but no RPC ever sets it — the claim path jumps straight to processing. It exists only so the watchdog reaper can also sweep never-claimed or legacy rows that were parked in dispatched. The proto mirror (TaskStatus) adds the usual proto3 zero value TASK_STATUS_UNSPECIFIED.

The columns that make it a queue (rather than just a history table):

Column	Why it exists
status	Lifecycle state; the claim index filters on it
tool_id	Which worker pool claims this row (FK to generation.tools)
claim_expires_at	The lease — how a live claim is held and how zombies are detected
retry_count / max_retries	Bounded retry (default max_retries = 3)
priority	Subscription-tier queue priority; higher is claimed sooner (default 0)
created_at	FIFO tie-breaker within a tool + priority band
reserved_token_amount / billing_idempotency_key	Durable billing context, read back to settle
w3c_traceparent	Parent span captured at enqueue, so the worker continues the same trace

Lifecycle

stateDiagram-v2
    [*] --> queued : QueueTask (INSERT)
    queued --> processing : ClaimNextTask (SKIP LOCKED, stamps lease)
    processing --> processing : HeartbeatTask extends lease
    processing --> processing : lease lapsed, reclaimed (retry_count++)
    processing --> completed : CompleteTask
    processing --> queued : FailTask, retries remain
    processing --> failed : FailTask, retries exhausted
    completed --> [*]
    failed --> [*]

1. Enqueue. core-gateway-consumer (consumer path) or core-mcp (agent path) runs its pricing + reserve pipeline, then calls QueueTask — a single INSERT … RETURNING. The row lands as queued. See the full request lifecycle.

2. Claim. A worker’s WorkerFleet calls ClaimNextTask(tool_id). This is one atomic UPDATE that flips the winning row to processing, stamps started_at, and sets the lease.

3. Work + heartbeat. The worker runs your tool’s process() and, on a background thread, calls HeartbeatTask to keep the lease alive.

4. Terminate. CompleteTask (with reported_usage for settlement) or FailTask. Then billing settles against the reservation — covered in the token ledger & money path.

Claiming: SKIP LOCKED, no double dispatch

ClaimNextTask is a single statement — an UPDATE whose target is chosen by a locked SELECT sub-query:

ClaimNextTask (excerpt)

UPDATE generation.tasks
SET status = 'processing',
    started_at = COALESCE(started_at, NOW()),
    claim_expires_at = NOW() + INTERVAL '2 minutes',
    retry_count = CASE WHEN status = 'processing' THEN retry_count + 1 ELSE retry_count END
WHERE id = (
  SELECT id FROM generation.tasks
  WHERE tool_id = NULLIF($1, '')::uuid
    AND retry_count < max_retries
    AND (status = 'queued' OR (status = 'processing' AND claim_expires_at < NOW()))
  ORDER BY status DESC, priority DESC, created_at ASC   -- zombies, then tier, then FIFO
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
RETURNING …;

Three things are doing the work here:

• FOR UPDATE SKIP LOCKED — two workers polling the same tool at the same instant never grab the same row. The loser skips the locked candidate and takes the next one. No external broker, no double dispatch.
• The partial index idx_gen_tasks_queue on (tool_id, status DESC, priority DESC, created_at ASC) WHERE status IN ('queued','processing') makes that ORDER BY … LIMIT 1 an index-ordered top-1 even under a deep, mixed-priority per-tool backlog.
• The ORDER BY — because the enum is declared queued, dispatched, processing, …, status DESC sorts processing (a reclaimable zombie) ahead of queued, so dead work is recovered first; then priority DESC honours the subscription tier; then created_at ASC is plain FIFO.

The retry_count = CASE … clause reads the pre-update status: reclaiming a zombie (it was already processing) burns one retry; a fresh queued claim does not.

Leases, heartbeats, and zombie reclaim

Ownership of a claimed task is a time-based lease, not a database lock held open across the whole (potentially minutes-long) generation. Holding a row lock that long would pin a Postgres connection and fall apart the moment a worker crashed.

• Claim stamps a lease. claim_expires_at = NOW() + INTERVAL '2 minutes'.
• HeartbeatTask extends it. WorkerFleet calls it every heartbeat_interval (default 60s) while the task is in flight — a bare UPDATE … SET claim_expires_at = NOW() + INTERVAL '2 minutes' WHERE id = … AND status = 'processing'. A 60s heartbeat against a 120s lease tolerates one missed beat before the lease is considered dead.
• Dead lease → reclaim. If a worker dies mid-task, nothing extends the lease. The next ClaimNextTask sees status = 'processing' AND claim_expires_at < NOW(), re-claims the zombie (ahead of fresh work, per the ordering above), and increments retry_count.

The lease is not a Postgres lock

The FOR UPDATE SKIP LOCKED lock exists only for the microseconds of the claim UPDATE. Durable ownership lives in the claim_expires_at timestamp, which is why a crashed worker’s task self-heals instead of stranding behind a held connection.

Retries and the unclaimable-row trap

FailTask(increment_retry=true) re-queues a task while attempts remain and only marks it terminal failed once retries are exhausted — the row flips back to queued so the same ClaimNextTask picks it up again.

max_retries = 0 makes a row unclaimable forever

ClaimNextTask’s predicate is retry_count < max_retries. A row with max_retries = 0 (e.g. a legacy row queued before the default was 3) can never be claimed — 0 < 0 is false — so it sits queued forever with no worker ever touching it, and a chat message bound to it hangs at “planning…” indefinitely. This is not fixed at claim time; it is swept by the watchdog reaper (below), which fails such rows so the UI can move on.

The watchdog reaper

A stuck task is not terminal, so it never shows up in the billing settlement sweep — it just loops the UI loader forever. ListStaleTasks is the backstop. It returns three populations, mirroring exactly what ClaimNextTask can and cannot still recover:

1. Dead zombies — processing, retries exhausted, claim_expires_at past its expiry by a grace window (processing_grace_seconds, default 120s). A worker died mid-task and no other worker will ever reclaim it.
2. Unclaimable queued/dispatched rows — retry_count >= max_retries (the max_retries = 0 case above).
3. Abandoned rows — never-claimed queued/dispatched older than a generous cap (queued_max_age_seconds, default 900s). The cap is deliberately generous so a brief worker outage self-heals via ClaimNextTask before it is reaped.

The consumer is the gateway’s background reconciler, app/grpc_server/reconciler.py, which runs every reconcile_interval_seconds (default 60s). Each sweep calls ListStaleTasks, then FailTask(increment_retry=false) to force each row terminal failed, and projects the now-failed state onto the chat message so the loader resolves. Once terminal, the row becomes eligible for the settlement sweep and its reservation is refunded (idempotently) on the next pass. See billing reconciliation for the money side.

Fast pickup: the task:wake pub/sub

Polling alone means a worker discovers a fresh task only on its next tick — a ~1–3s idle gap. To close it, the gateway pushes a best-effort wake signal, with polling kept as the correctness floor.

• Producer. After QueueTask commits, GenerationPipeline (app/enforcement/pipeline.py) publishes the tool_id to the Redis channel {env}:common:task:wake (commit-before-publish, wrapped in try/except so a Redis hiccup never fails the request).
• Consumer. Each WorkerFleet (src/soundverse/tools/fleet.py) runs a daemon subscriber that filters the payload against the tools it hosts and sets a wake event. The poll loop clears-before-sweep (no lost wakeup), fast-drains while it keeps claiming, backs off 0.25s when at concurrency capacity, and otherwise parks on the wake event with a jittered fallback of poll_interval (default 2.0s).

The wake is best-effort — polling is the floor

Pub/sub is lossy and fire-and-forget: a dropped wake is not an error, it just means the task waits for the next poll tick. Never treat delivery as guaranteed. This is what makes the push deploy-order-independent — you can ship the producer or the consumer first and the system still drains correctly, only slower for the transitional window.

Redis wiring gotchas that silently drop wakes

Two alignment traps, both about the fleet-wide Redis contract:

• Var name. Workers, core-mcp, and core-database all read REDIS_ADDR; the gateway historically read REDIS_URL and sat on its localhost default, so wakes never reached the shared Redis. It now reads REDIS_ADDR (with REDIS_URL as an alias).
• DB index + environment. Pub/sub is scoped per Redis DB index — and the index lives in the path of the REDIS_ADDR URL, so all services must point at the same one. The channel is also namespaced by deploy ENVIRONMENT ({env}:common:task:wake), so a worker must run under the same environment as the gateway to hear its wakes.

See the Redis key conventions for the {env}:{namespace}: contract.

Not the queue: the live-events stream

Don’t confuse the task:wake channel (one throwaway pub/sub ping meaning “a task is queued”) with the per-task Redis Stream {env}:common:task:{id}:events. The worker XADDs progress, streaming-URL, partial-output, and terminal events to that stream via TaskContext; core-mcp XREADs and relays them as MCP progress notifications. The queue lives in Postgres; only these two signalling side-channels live in Redis.

Related

• Request lifecycle — where QueueTask and ClaimNextTask sit in the end-to-end path
• The soundverse-py SDK — WorkerFleet: claim, heartbeat, concurrency lanes, drain
• Token ledger & the money path — reserve → settle against the terminal task
• Billing reconciliation — the unsettled + stale-task sweeps in detail
• How the DB codegen plugin works — how each SQL-annotated task RPC becomes a Go handler