.. This file is part of PEPSI. Copyright (C) 2026 Pepsi contributors PEPSI is free software; you can redistribute it and/or modify it under the terms of the GNU Affero General Public License as published by the Free Software Foundation; either version 3, or (at your option) any later version. ============== pepsi-dispatch ============== *The single long-lived coordinator that drives messages through the stages.* 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: :manpage:`pepsi-dispatch(1)`. Features ======== * **Event-driven, batched claiming.** ``LISTEN``\ s 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 :doc:`pepsi-httpd`'s ``/metrics``. 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-]`` section. See :manpage:`pepsi.conf(5)` and :doc:`../configuration`. See also ======== :doc:`../architecture`, :doc:`pepsi-queue`, :doc:`pepsi-ingress`, :manpage:`pepsi-dispatch(1)`.