.. This file is part of PEPSI. Copyright (C) 2026 GNUnet e.V. 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, as ``pepsi-dispatch serve`` (the one subcommand; the global ``-c``/``-L``/``-v`` flags come before it). It claims newly queued rows and feeds each to an idle **worker process** of the row's stage, requeues paused messages when their retry time is due, and recovers from crashes and timeouts. It processes no message content itself. Started as ``root`` it drops to the ``pepsi`` service account, which the workers inherit. Reference: :manpage:`pepsi-dispatch(1)`. Features ======== * **Event-driven, batched claiming.** ``LISTEN``\ s on the ``workqueue`` and ``config_changed`` channels 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 workqueue_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. * **Advancing is the stage's write.** A stage that advances a message sets the row ``pending`` at the next stage itself, in the same update that moves ``stage``; the dispatcher never rewrites a row on success. The worker's status report brings the coordinator round to its next claim, which picks the row up for the next stage's pool (so an advance needs no notification). A worker may also *fuse* a fast successor into the same pass (see :manpage:`pepsi-dispatch(1)`, *Stage fusion*). * **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`` (before the ``worker`` subcommand) so they load the same configuration. * **Hot reload of the pipeline.** On a ``config_changed`` notification it rebuilds its own stage table from the database overlay and retires every worker — idle ones at once, busy ones when their current message reports back — so a ``[stage-*]`` change takes effect without a restart. A reloaded graph that does not parse ends the process rather than routing by a pipeline the workers no longer share. Its own ``[pepsi-dispatch]`` options are *not* reloaded. * **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), ``DB_POOL_SIZE`` (default 8) and ``CONFIG_FILE`` (no default; without it the workers fall back to the default configuration search). 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)`.