Queues

Each actor instance has a persisted queue in Postgres (actor_queues). Fan-out enqueues matched stream events; the scheduler claims and dispatches them to Deno process().

Queue row schema

actor_queues (
  instance_id,
  event_id,
  event,          -- full StreamEvent JSON
  status,         -- pending | processing
  created_at,
  retry_count,
  tenant_id,
  priority,
  group_id
)

Claim order

When the scheduler claims the next event, Postgres selects by queue policy mode:

LIMIT 1
FOR UPDATE SKIP LOCKED

There is no mode: "priority" queue policy — priority is a per-event field set at append or fan-out time.

Stale processing reclaim

If a worker crashes mid-process(), rows stay in processing until reclaimed. The scheduler periodically resets rows where claimed_at is older than YOKE_PROCESSING_TIMEOUT_SECS (default 300) back to pending.

Queue policy modes

Queue policy is set on the actor definition and serialized on the instance record:

type QueuePolicy =
  | { mode: "fifo"; concurrency: number }
  | { mode: "lifo"; concurrency: number }
  | { mode: "random"; concurrency: number }
  | { mode: "custom"; select: (queue) => event | events[]; concurrency: number };

concurrency controls parallel process() invocations per group_id. Infinity is valid.

Custom selectors after restart

Custom select functions are not serializable. Provide queueSelectFromConfig on the definition:

queueSelectFromConfig: (config) => (events) =>
  events.find((e) => e.type === config.preferredType) ?? events[0],

Processing lifecycle

pending → processing → (deleted on success)
                    → pending (on throw, retry)
                    → dead_letters (on saturation / max retries)

1. Scheduler claims event (status = processing)
2. dispatch_process RPC → Deno process(ctx, event)
3. Success → row deleted
4. Throw → status reverted to pending, retry_count incremented

There is no early-ack. The event stays processing for the entire process() duration.

Concurrency groups

groupId on events scopes concurrency slots:

await ctx.append(
  { type: "work.item", payload: { id: "a" } },
  { groupId: "batch-1" },
);
await ctx.append(
  { type: "work.item", payload: { id: "b" } },
  { groupId: "batch-2" },
);

With concurrency: 1, both groups process in parallel. Within a group, only one process() runs at a time (per concurrency setting).

The scheduler calls list_active_groups to find groups with pending work, then claims per group independently.

Dead letters

When queues saturate or retries exhaust, events move to dead_letters:

Metric: yoke_queue_dead_letters_total.

Backpressure

Fan-out always enqueues — it does not block on queue depth. Very deep queues increase latency for that instance but do not affect the stream or other instances.

Design patterns for high-volume instances:

• Increase concurrency
• Split work across instances with narrower filters
• Use priority to prefer urgent events
• Dead-letter policy for poison messages

Queue depth metrics

yoke_queue_depth gauge with instance_id label — per-instance pending + processing count.

Implementation

The Postgres queue backend (PgActorQueue) implements these key methods:

Related