23. pepsi-dispatch

The single long-lived coordinator that drives messages through the stages.

23.1. Role

pepsi-dispatch is the only process that starts stage programs. Run exactly one per system. It claims newly queued rows and feeds each to an idle worker process of the row’s stage, requeues advanced messages for the next stage, and recovers from crashes and timeouts. It processes no message content itself. Reference: pepsi-dispatch(1).

23.2. Features

• Event-driven, batched claiming. LISTENs on the ingress channel with a periodic POLL_INTERVAL heartbeat. On any wake it drains its pending internal events (so per-stage capacities are current) and then claims work for every stage with spare capacity in one UPDATE … RETURNING — each stage up to its own remaining capacity (a per-stage LATERAL … ORDER BY ingress_id LIMIT cap that stops at the cap via the partial pending index). With exactly one dispatcher claiming serially as the sole pending``→``running writer, the claim needs no FOR UPDATE SKIP LOCKED.

• Persistent, pipelined workers. Each stage runs PROGRAM worker (found on PATH unless absolute) as long-lived processes that connect to the database once and read message ids from standard input, answering with a one-line status (0 on success) per message, in order. A worker is handed up to QUEUE_LIMIT ids at once (written to its stdin in a single buffer); it processes them strictly serially, but the next id is already queued, removing a coordinator round-trip per message. A stage’s in-flight capacity is therefore QUEUE_LIMIT × PARALLELISM, and because the worker count is unchanged it costs no extra database connections.

• Elastic per-stage pools. Workers are started on demand up to a stage’s PARALLELISM; work is packed onto the fewest workers, and surplus workers are stopped after WORKER_IDLE_TIMEOUT — so a quiet stage collapses toward zero workers and a busy one grows to its cap. After MAX_MESSAGES a worker recycles its child (once its pipeline drains); a stage whose PROGRAM cannot start is held off with a short cooldown instead of spinning.

• Requeue on advance. When a worker reports success leaving the row running at a new stage (it advanced), the dispatcher requeues it to pending so the next stage’s pool picks it up (no in-process chaining).

• Retry scheduling. Sleeps until the earliest paused row’s timeout (or a due worker reap / spawn cooldown / the heartbeat), then flips due rows back to pending and re-scans.

• Crash & watchdog recovery. A worker that reports non-zero fails that one message and survives; a crashed child fails its head-of-line message, and a head-of-line message unanswered within MAX_RUNTIME (timed from when it reached the head) → timeout. In the crash and timeout cases the worker’s other pipelined messages never ran and are requeued to pending rather than lost. Reasons are written to state.dispatch_error under a status = 'running' guard so a stage that already transitioned the row is never clobbered.

• Start-up / shutdown safety. Resets orphaned running rows (from a previous dispatcher) to pending on start; on SIGINT/SIGTERM stops workers and resets their (and any claimed-but-unassigned) rows.

• Isolation. At most one writer ever touches a given queue row, so the shared pool runs READ COMMITTED; queue operations are still wrapped in a retry helper that re-runs a transient 40001/40P01 failure as cheap insurance.

• Config propagation. Passes CONFIG_FILE to workers via -c so they load the same configuration.

• Statistics. Accumulates per-stage throughput, kills/timeouts, crashes and processing time plus the global stage/message totals in memory, and flushes the deltas to the stage_stats/dispatch_stats tables in a single transaction every STATS_INTERVAL, whenever the pipeline goes idle (at most once a second, so a finished burst’s numbers do not wait for the next tick), and on shutdown; nothing is written when there are no deltas. It is the only writer of those tables — stages fused into another stage’s worker pass are reported back on that pass’s status line instead of being written by the worker, which keeps every worker of a stage from contending for that stage’s single row. These are exported by pepsi-httpd’s /metrics.

23.3. Configuration

[pepsi-dispatch]: MAX_RUNTIME (default 300 s), WORKER_IDLE_TIMEOUT (default 5 s), POLL_INTERVAL (default 30 s), STATS_INTERVAL (default 60 s), CONFIG_FILE. Per-stage worker pools are sized by PARALLELISM (default 4), QUEUE_LIMIT (default 4, the messages pipelined to one worker at once) and MAX_MESSAGES (default 1000) in each [stage-<name>] section. See pepsi.conf(5) and Configuration.

23.4. See also

Architecture, pepsi-queue, pepsi-ingress, pepsi-dispatch(1).