70.1.2. pepsi-dispatch

drive messages through the stage pipeline

Manual section:

1

70.1.2.1.1. Name

pepsi-dispatch - claim pending messages and run their stage programs.

70.1.2.1.2. Synopsis

pepsi-dispatch [GLOBAL-OPTIONS] serve

70.1.2.1.3. Description

pepsi-dispatch is the long-lived dispatcher that advances messages through the stage pipeline (see pepsi.conf(5)). It claims pending rows from pepsi.ingress, sets each to running, looks up the message’s stage in its [stage-<stage>] section, and hands the message id to a worker process of that stage. Each stage runs as a pool of persistent workers started as PROGRAM worker (the program is found on the PATH unless PROGRAM is an absolute path); a worker connects to the database once and then reads message ids from its standard input and writes one status line per message to its standard output (0 on success — the value the process exit code carried in earlier releases), in input order. A status may be followed by a space and a JSON report of the stages that were fused into that pass, [["<stage>",<microseconds>], …], which is how per-stage statistics stay whole without the workers writing them (see Statistics below). Only an unreadable status is fatal — it desynchronises the positional protocol, so the dispatcher tears the worker down; an unreadable report is dropped with a warning, so a worker and a dispatcher from different builds lose statistics for the length of a restart window rather than messages. Run exactly one dispatcher per system; it is the only process that starts stage programs (though many workers then run concurrently against the shared database). New work is picked up from a LISTEN on the ingress channel, with a periodic safety-net heartbeat (POLL_INTERVAL).

Notifications on that channel are issued explicitly, and only by writers outside the dispatcher’s own loop: pepsi-ingress(1) (once per burst of accepted messages, not once per message), the ingress_inject and ingress_resume SQL functions, pepsi-keydisc(1) releasing parked mail, pepsi-failure-bouncer(1) and the pepsi-queue(1) repair commands. A stage advancing a message deliberately does not notify: the worker reports the finished message on its standard output, and the dispatcher answers that with the same full claim a notification would have triggered, so the notification would carry no information. Earlier releases used a table trigger that fired for every row becoming pending; PostgreSQL holds a database-wide lock from the moment a transaction queues a notification until it commits, so that made every stage hop serialise the whole database’s commits — see the Performance chapter of the manual.

Batched claiming. On any wake (a notification, a freed or dead worker, the poll heartbeat, a listener (re)connect) the dispatcher first drains every pending internal event so its per-stage capacities are current, then runs one UPDATE … RETURNING that claims, for every stage with spare capacity at once, up to that stage’s own remaining capacity (pending``→``running, bounded per-stage by a LATERAL subquery whose ORDER BY ingress_id LIMIT cap stops scanning each stage’s backlog at its cap, driven by the partial index on (stage, ingress_id) WHERE status='pending'). The notification carries an empty payload — a notification is simply a wake, and a burst of them collapses into one batched claim (PostgreSQL also coalesces identical queued notifications).

Pipelined, elastic worker pools. Each stage’s pool grows on demand up to its PARALLELISM (default 4) and shrinks when idle: no worker runs until the stage sees traffic; when a stage has work and spare capacity the dispatcher fills an existing worker’s free slot or starts a new worker; work is packed onto the fewest workers so a surplus goes cold and is stopped after WORKER_IDLE_TIMEOUT (default 5 s). Under steady light load a stage therefore hovers around a single worker. A worker is handed up to QUEUE_LIMIT (default 4) message ids at once — written to its standard input in a single buffer — rather than one-at-a-time; it still processes them strictly serially, but the next id is already queued, so completing a message frees it without a coordinator round-trip first. A stage’s total in-flight capacity is thus QUEUE_LIMIT × PARALLELISM. Because the worker count (and so the database-connection count) is unchanged, raising QUEUE_LIMIT trades a little head-of-line latency for throughput without spending more connections. After MAX_MESSAGES (default 1000) a worker recycles its child — once its pipeline has drained — with a fresh process to bound memory growth.

Batched body-free stages. A worker for a stage that loads neither the header block nor the body (the metadata-only stages — srs, if, the whitelist stages, discard, aliases …) processes its pipelined ids in one batch rather than one at a time: it greedily takes every id already buffered on its standard input (up to QUEUE_LIMIT, stopping the instant a read would block, so a lone id is still handled immediately), loads them all in a single SELECT … WHERE ingress_id = ANY(...), runs each stage body, then commits the advances/fails/requeues in one arrayed UPDATE … FROM jsonb_to_recordset(...) per outcome shape (typically one). It still writes one status line per id, in order, so the worker protocol is unchanged. When the queue is full this cuts a body-free stage’s database round-trips from two per message to roughly two per QUEUE_LIMIT messages. Stages that load the headers/body keep the one-id-at-a-time path (their large columns do not array cheaply).

Advancing. A stage advances a message by setting its stage column to the next stage and the row back to pending in one update; the dispatcher never rewrites a row on success. Claiming (pending``→``running) is the dispatcher’s only write to a row’s status for forward progress, and it is the only writer that performs it. When a worker reports success (status 0) the dispatcher re-scans for work — which is precisely why an advance issues no notification of its own — so the next stage’s pool claims the now pending row on the following scheduling round (unlike earlier releases, an advanced message is no longer chained inside the same process). The message is done when the worker deletes the row, paused when it leaves it paused, or terminal when failed/timeout. A stage commits its outcome in a single round-trip — one UPDATE/DELETE, or one stored function that fans work out server-side (a recipient split, or a finish/pause that also spawns bounce, delay or success DSNs from arrays of clones) — never a multi-statement client transaction.

Failures and retries. A worker that reports a non-zero status leaves that message failed with the reason in state and stays alive for the next id. A worker whose child closes its output (a crash) fails its head-of-line message and is torn down; a child whose head-of-line message is not answered within MAX_RUNTIME (real time) is killed and that message set to timeout. In both of those cases the other messages already pipelined to that worker never ran, so they are reset to pending and re-dispatched (a single stuck or poison message therefore does not strand its in-flight siblings). The MAX_RUNTIME clock is per message and starts when a message reaches the head of the worker’s queue, so a message waiting behind a slow sibling is not charged for the wait. When a stage leaves a message paused it records, in timeout, when the message should be retried; the dispatcher sleeps until the earliest such time, flips the due rows back to pending, and re-scans for work. At start-up it resets any leftover running rows (orphaned by a previous dispatcher) back to pending. Because there is exactly one dispatcher and its coordinator claims serially — the sole writer that sets pending``→``running — the batched per-stage claim needs no FOR UPDATE SKIP LOCKED. Because at most one writer ever touches a given row, the shared database connection runs READ COMMITTED rather than SERIALIZABLE (queue operations are still routed through a retry helper as cheap insurance).

Database-overload backpressure. A worker that cannot reach the database because its connection limit is exhausted (PostgreSQL too_many_connections, or its pool times out acquiring a connection) reports a distinct status (EX_TEMPFAIL, 75) rather than the generic failure code — this is infrastructure pressure, not a defect in the message. The dispatcher then does not fail the message: it requeues it to pending for a later retry, and to shed connections it temporarily reduces the parallelism of the stage currently running the most worker processes (the largest contributor to the pressure, which need not be the stage that reported the error) — halving its cap, floor 1, for five minutes. Repeated reports ratchet that stage down further and refresh the window; existing workers drain via the idle reaper and the stage ramps back up once the window passes. The remedy for sustained pressure is to raise PostgreSQL’s max_connections or lower the stages’ PARALLELISM so that the sum of every component’s pool fits the server (see the [pepsi-postgres] connection-budget note in pepsi.conf(5)).

Per-address settings. Before a worker runs a stage’s logic it consults the pepsi.settings table (see pepsi-settings(1)) for the message’s correspondent — the envelope sender of a state.local_origin message, otherwise its recipients — and layers any per-address overrides onto the stage’s [stage-<name>] options. The overrides are fetched with the message row in the same query (a correlated subquery on the relevant addresses), so loading a message and its settings is a single round-trip. When the recipients of one inbound message resolve to different overrides for the stage about to run, the worker splits the message lazily: in one round-trip (the ingress_split function) it groups the recipients by their effective override, keeps the first group on the current row, and fans each remaining group out as a new pending row at the same stage (the trigger wakes the dispatcher for each). Each row then runs the stage under its own settings. A message whose recipients never diverge is never split.

Stage fusion. Many stages are very fast and need only the message metadata (the envelope, the extracted From:/Subject: and the state JSON), not the header block or body. When such a stage advances to a successor that needs no more data than is already loaded, routing the row back through the database and the dispatcher is pure overhead. Stage fusion removes it: instead of writing the row pending at the next stage and waiting for it to be claimed, the worker runs the successor’s body in the same process, reusing the row it already loaded. A whole chain of fused stages is one SELECT at the front, the stages’ own work, and a single terminal write, reported to the dispatcher as one completion.

A successor is fused only when all hold: stage fusion is enabled globally ([pepsi] ALLOW_FUSION, on by default); the successor’s section is marked FUSION = yes (the default for the fast metadata-only stages — pepsi-stage-if(1), pepsi-stage-discard(1), pepsi-stage-srs(1), pepsi-stage-check-whitelist(1), pepsi-stage-auto-whitelist(1), pepsi-stage-block-language(1), pepsi-stage-edit-settings(1)); the successor’s PROGRAM is folded into the same unified pepsi binary (so it can run in-process — fusion is therefore inert in a per-program multibin build); and the successor needs no message column the predecessor did not load. Any miss makes the advance a normal dispatched hop, so fusion never changes a message’s outcome — only whether the hand-off touches the database. A stage may additionally apply a data-dependent gate that refuses fusion for some messages: pepsi-stage-edit-settings(1) loads the full body, yet it does nothing to any message that is not a settings control message, so it fuses through every non-control message on metadata alone and declines fusion only for a genuine control message — which is then committed and dispatched normally so the worker loads its body. Per-address settings are still applied to each fused stage (their overrides were already fetched with the row), and a divergent recipient split still happens where needed. A misconfigured stage cycle is bounded by a fusion-depth cap, after which the advance is committed normally. Fused stages are still counted individually in pepsi.stage_stats, so per-stage statistics stay accurate — the worker reports them on the pass’s status line and the dispatcher folds them in (see Statistics); only the per-stage cost benchmark turns fusion off (with ALLOW_FUSION = no) so it can time each stage as its own dispatched worker.

Statistics. The cumulative counters in pepsi.stage_stats and pepsi.dispatch_stats are accumulated in memory by the dispatcher and written in a single transaction: every STATS_INTERVAL, whenever the pipeline goes idle (so a burst’s numbers are visible as soon as it ends rather than up to an interval later — rate-limited to at most one such flush a second), and on shutdown. An idle dispatcher performs no database work at all, because a flush with no accumulated deltas writes nothing.

The dispatcher is deliberately the only writer of these tables, which is why a fused hop is reported to it on the worker’s status line rather than written by the worker: stage_stats holds one row per stage, so a counter written per message would make every worker of a stage serialise on that one row. Carrying the figures on a line the worker already writes removes the write entirely. The trade is the usual one for statistics: deltas not yet flushed are lost if the dispatcher dies, which is why these counters are documented as best-effort. stages_executed counts worker passes, so a fused chain counts once however many stages it covers.

pepsi-dispatch does not itself process messages — it only runs the stage workers, which record their own outcome on the row.

70.1.2.1.4. Commands

serve

Run the dispatcher until interrupted. Requires that the schema has been installed with pepsi-setup(1).

70.1.2.1.5. Global Options

-c FILE, –config FILE

Read the configuration from FILE instead of searching the default locations. Set CONFIG_FILE in [pepsi-dispatch] to the same path so spawned stage programs inherit it (see pepsi.conf(5)).

-L LOGLEVEL, –log LOGLEVEL

Set the logging verbosity (default info).

-v, –verbose

Show log messages from all sources.

-h, –help; -V, –version

Print a usage summary / the version and exit.

70.1.2.1.6. Signals

SIGINT, SIGTERM

Initiate shutdown: stop claiming new work, stop every worker (killing its child), reset any in-flight or claimed-but-unassigned row back to pending, and exit.

70.1.2.1.7. Exit Status

0

Clean shutdown.

1

An error occurred (for example a malformed configuration file or a failed database connection). The reason is written to the log.

70.1.2.1.8. Examples

Run the dispatcher:

pepsi-dispatch -c /etc/pepsi/pepsi.conf serve

70.1.2.1.9. See Also

pepsi-config(1), pepsi.conf(5), pepsi-stage-srs(1), pepsi-stage-bounce(1), pepsi-stage-dkim-sign(1), pepsi-stage-relay-to-smarthost(1), pepsi-stage-relay-to-internet(1), pepsi-settings(1), pepsi-queue(1), pepsi-setup(1)

70.1.2.1.10. Bugs

Report bugs to the Pepsi issue tracker.