.. 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. ====================== Extending the Pipeline ====================== Because the pipeline is a state machine over a single table, adding a processing step does not touch ingress, the dispatcher or any other stage: you write a new **stage program** and splice its ``[stage-]`` section into the graph. This chapter is the recipe. A stage program is a persistent worker the dispatcher starts as ``PROGRAM worker``; it reads one ``ingress_id`` per line of standard input and writes one status line per message to standard output (``0`` on success). For each id it must: load the row, do its work, and call **exactly one** terminal helper. The same body also runs once as ``PROGRAM `` for manual operation and tests. Everything shared — the stdin/stdout worker loop, opening the pool once, loading the row, verifying the section — is provided by ``pepsi_common::stage`` (``run_worker`` / ``run_oneshot``). **Standard output is the worker protocol channel, so a stage body must never write to it** (log to standard error). When a stage is the right tool ============================== Write a stage when you want to inspect or transform a message in flight, or make a routing decision, e.g.: * a content filter (spam/virus) that ``pause``\ s, ``fail``\ s or advances; * a header rewriter or archival/audit hook; * an alternative delivery method (a new transport); * an integration that calls out to another service per message. If you only need a different *wiring* of existing stages, you do not need code at all — just new ``[stage-*]`` sections. Anatomy of a stage crate ======================== Follow the existing stages (``pepsi-stage-srs`` is the smallest). A stage is a library crate plus a thin binary in the workspace root: #. **Library crate** ``pepsi-stage-`` with: * ``constants.rs`` — a ``CONFIG_SOURCE`` and a ``PROGRAM`` string (the binary name operators put in ``PROGRAM =``). * ``config.rs`` *(optional)* — a struct parsed from the stage's ``Section`` for any program-specific options. * ``lib.rs`` — the stage body ``async fn body(ctx: StageContext<'_>) -> anyhow::Result<()>``, a ``LOAD`` constant, and the two thin entry points ``run`` (``run_oneshot``) and ``worker`` (``run_worker``). #. **Binary** ``src/bin/pepsi-stage-.rs`` that calls ``taler_main``, exposes a ``worker`` subcommand delegating to the library's ``worker``, and runs the library's ``run`` for a bare ```` argument. #. Register the crate in the workspace ``Cargo.toml`` and add the binary to the ``Makefile``'s ``BINARIES`` list. The body ======== A stage body is uniform. The two entry points just name the body and how much of the message to load; the body acts and terminates:: // Load only what you touch. Metadata loads neither the header block nor the // (possibly large) body; Headers adds the header block; Full adds both — use // Full only if you must hash or transmit the whole message. const LOAD: stage::Load = stage::Load::Metadata; /// One-shot: process a single id (manual operation, tests). pub async fn run(message_id: i64, cfg: &Config) -> anyhow::Result<()> { stage::run_oneshot(PROGRAM, message_id, cfg, LOAD, act).await } /// Persistent dispatcher worker: read ids from stdin until EOF. pub async fn worker(cfg: &Config) -> anyhow::Result<()> { stage::run_worker(PROGRAM, cfg, LOAD, act).await } async fn act(ctx: StageContext<'_>) -> anyhow::Result<()> { let id = ctx.message.ingress_id; // Read program options from this stage's own section: let mycfg = MyCfg::parse(&ctx.section())?; // Read inputs you need from state / columns (ctx.cfg is the Config): let dsn = DsnParams::from_state(ctx.message.state_value(), 0); // ... decide what to do (never println! — stdout is the worker channel) ... // Terminate with exactly ONE of the helpers (see below). ctx.advance().await } Key pieces of the ``StageContext`` API: * ``ctx.message`` — the loaded row: ``ingress_id``, ``mail_from``, ``rcpt_to``, ``from_header``, ``subject``, optional ``headers``, optional ``body``, ``state``, ``age_secs``, and helpers ``is_bounce()`` and ``state_value()``. * ``ctx.section()`` — the stage's ``[stage-]`` section for your options. * ``ctx.headers()`` — the header block (requires ``Load::Headers`` or ``Load::Full``). * ``ctx.raw_message()`` — the reassembled message (requires ``Load::Full``). * ``ctx.require_next_stage()`` — the configured ``NEXT_STAGE``. * **Content setters** — to *rewrite* the message, never issue SQL: record the change with ``set_mail_from``, ``set_recipients``, ``set_from_header``, ``set_subject``, ``set_headers``, ``set_body``, ``set_state`` (replace), ``merge_state`` (shallow ``||`` merge), ``clear_state`` (set ``NULL``), or ``set_auth_verdict(key, val)`` (set ``state.auth.``). The change is folded into the in-memory row and written by the worker together with your terminal transition in **one** ``UPDATE``. A content rewrite must therefore end in an advancing/failing terminal (``advance``/``reroute``/``fail``/``set_pending``), never a fan-out one (``finish``/``pause``/``split_off_recipients``), which would not persist it. * ``ctx.telemetry_update("")`` — record that your feature was exercised (non-blocking, infallible, and a no-op unless the operator opted in by turning ``[pepsi] SHARE_TELEMETRY`` on, which is not the default). Call it where the feature actually does its work, with the **stable key** you registered for it (see `Keeping the feature-stability table current`_ below). Terminal helpers (choose exactly one) ===================================== The terminal helper you call **is** the contract with the dispatcher: .. list-table:: :header-rows: 1 :widths: 30 70 * - Helper - Effect * - ``advance()`` / ``advance_to(s)`` - Move to ``NEXT_STAGE`` (or ``s``); stays ``running``. The dispatcher requeues the row to ``pending`` for the next stage's worker pool. * - ``reroute(stage, state)`` - Move to ``stage`` merging ``state``; used to hand off to a bounce stage. * - ``pause(state, secs)`` - ``paused`` + a retry ``timeout``; the dispatcher re-queues it later. * - ``fail(state)`` - Terminal ``failed``, left for an operator. * - ``finish()`` - ``DELETE`` the row (message consumed). * - ``advance_or_finish()`` - Advance if a ``NEXT_STAGE`` exists, else finish. * - ``complete_success(dsn, diag, rcpt)`` - Delivery-style success: emit a positive DSN when configured, else ``advance_or_finish``. Rules the stages live by ======================== To stay consistent with the rest of the pipeline, honour these invariants: * **One SELECT, one UPDATE.** Pick the right ``Load``; record content changes via the setters and end with one terminal helper. The worker commits every change and the transition in a single ``UPDATE`` built from exactly the columns you touched — don't issue your own SQL or add extra round-trips. * **Preserve ``state.dsn``** (and other keys) — use ``merge_state`` (not ``set_state``), so don't overwrite ``state`` wholesale unless you are intentionally minting a new message (only the bounce stage does that, via ``clear_state``). * **Never bounce a bounce.** Check ``ctx.message.is_bounce()`` before producing a notification; a null-sender message is dropped, not re-bounced (RFC 5321 §6.1). * **Honour DSN ``NOTIFY``.** If you originate notifications, gate them on the recipient's ``NOTIFY`` via ``pepsi_common::dsn`` (failure by default; success/delay only on explicit request). * **Fail-open where a stuck message is worse than an imperfect one** (the ARC and DKIM-sign stages advance with a warning rather than blocking) — but only when that is safe for your step. * **Be idempotent / restart-safe.** A stage may be retried after a crash; design so re-running on the same row is safe (the dispatcher resets orphaned ``running`` rows to ``pending``). Wiring it in ============ Build and install the new binary, then add a stage section and point a neighbour at it:: [stage-spamfilter] PROGRAM = pepsi-stage-spamfilter NEXT_STAGE = deliver # ... your program's own options ... [stage-init] PROGRAM = pepsi-stage-arc NEXT_STAGE = spamfilter # was: deliver Run ``pepsi-setup ... check`` (and ``pepsi-config ... dump``) to validate the graph — it verifies that ``[stage-init]`` exists and every ``NEXT_STAGE``/``BOUNCE_STAGE`` resolves. No ingress or dispatcher change is needed; the next message simply flows through your new stage. Testing a stage =============== Follow the project's testing constraints: never send real mail to third-party MX (use reserved/``.invalid`` domains), and remember the dev uid cannot bind ports 25/53. Unit-test the pure logic in your library; for an end-to-end check, run a throwaway PostgreSQL, install the schema with ``pepsi-setup``, insert a row, and invoke the binary on its id directly (``pepsi-stage- -c test.conf``) — the global ``-c`` flag must follow the message-id/subcommand. .. _adding-an-mta-importer: Adding support for migrating from another MTA ============================================= ``pepsi-setup`` can import an existing mail server's configuration as the wizard's answer defaults (see :doc:`installation` and :doc:`programs/pepsi-setup`). Five servers are supported out of the box — Postfix, Exim, Sendmail, qmail and Stalwart — and adding a sixth is one new file plus one line in a registry. An importer implements ``MtaImporter`` (``pepsi-setup/src/import/mod.rs``):: pub trait MtaImporter: Sync { fn id(&self) -> &'static str; // "postfix" fn label(&self) -> &'static str; // "Postfix" fn detect(&self, probe: &Probe) -> Option; fn import(&self, probe: &Probe, found: &Detected) -> Imported; } Four rules make the difference between an importer that helps and one that misleads: #. **It cannot fail.** ``import`` returns ``Imported``, not ``Result``. Every unreadable file, unparseable line, unrecognised directive and unsupported feature is recorded as a ``Warning`` on the model — with the ``file:line`` it came from — and the partially-filled model is returned anyway. A migration that understood nine settings out of ten is still worth having, and no configuration file on anyone's disk may be able to abort setup. #. **Every directive is accounted for.** Consume it, list it in the module's ``IGNORED`` table of genuinely irrelevant knobs, or report it with ``Imported::unknown``. Silence about a setting the operator relies on is the failure mode this feature exists to prevent. For features Pepsi has no equivalent for, use ``Imported::unsupported`` and say in one sentence what replaces it (usually a stage). #. **All filesystem access goes through** ``Probe``. It handles size limits, non-UTF-8 bytes and permissions, and — because it can be rooted at a directory — it is what lets the importer be unit-tested against a fixture tree under ``pepsi-setup/tests/fixtures//`` with no access to the real system. #. **Routing tables are not classified by the importer.** Parse them into ``RawAlias { key, targets, origin }`` values (``import::aliases`` has parsers for the two universal dialects) and let ``import::aliases::convert`` decide what Pepsi can express. It is the single place that knows Pepsi's alias-key grammar, the qualification styles, and how to report a target — a ``|command``, a file, a broken ``:include:`` — that has no equivalent. Fill in what you learned on ``Imported``: identity, domains, direction, smarthost (including credentials, always recording ``password_source``), local delivery, aliases, submission identities, and any option in the ``crate::advanced`` registry via ``set_advanced`` — the shared code turns all of it into wizard defaults, converted map files and the migration report. Two traps worth naming: convert durations with ``text::pepsi_duration`` (Pepsi's duration parser rejects the calendar units ``d`` and ``w``, so ``5d`` must become ``120 h``), and drop loopback networks when importing a trusted-network list (they have caused a mail loop before). Finally, add the importer to ``import::registry()`` and give it a row in ``contrib/feature-registry.tsv``. Testing an importer ------------------- Every fixture tree lives at ``pepsi-setup/tests/fixtures///`` and **is** a probe root: the directory stands in for ``/``, so ``pepsi-setup import --root `` reproduces by hand exactly what the tests run. Give a new importer at least three: a realistic installation, a second layout that server supports (a split configuration, a generated file, a relay-only host), and a *hostile* one — truncated lines, unterminated quoting, binary noise, a directory where a file belongs, an unreadable include. Any tree whose name contains ``hostile`` is required to produce warnings. Three suites then apply, and a new importer is covered by two of them the moment its fixtures land: ``pepsi-setup/tests/import_pipeline.rs`` Enumerates every fixture tree in the repository and holds all importers to the shared invariants: only the owning importer claims a tree, every warning carries a subject, an explanation and a file-naming origin, the converted alias map loads under all three qualification styles with ``converted + dropped`` equal to the number of source entries, seeded answers are never empty, and the rendered report accounts for every warning without emitting control characters. ``pepsi-setup/tests/import_.rs`` The per-server suite: what this importer understands, asserted end to end through the public API. ``tests/common/mod.rs`` provides the ``Case`` helper (``Case::load("postfix", "postfix/basic")``) with ``warned``/``not_warned`` assertions — ``not_warned`` is the one that proves a directive was *consumed* rather than swept into the UNKNOWN pile. ``tests/cli_import.rs`` Runs the real binary: the ``import`` preview, the chooser when several mail servers are installed, and a full ``--wizard --import`` migration whose generated configuration must pass Pepsi's own validation — the wizard refuses to write one that does not. Integrating an external credential-refresh service ================================================== Some stages authenticate to an upstream with a **short-lived credential** that must be rotated out of band. The reference case is the smarthost relay (:doc:`programs/pepsi-stage-relay-to-smarthost`) with ``AUTH = oauth``: it reads a SASL bearer token, fresh on every delivery, from a ``TOKEN_FILE``; that OAuth access token expires roughly hourly. Pepsi ships :manpage:`pepsi-helper-token-refresh(1)` to keep the file current, but that is only **one implementation of a contract** — you can drop in your own refresher (a cron job hitting a token endpoint, a cloud-provider agent, a secrets-manager sidecar) as long as it honours the same contract. This section documents it so you can. Why a separate service, not a stage ----------------------------------- Refreshing is not per-message work, it needs the provider's *client secret* (which the stage must never see), and it should keep running even when no mail is flowing. So it is a standalone program that runs as its **own, more-trusted user**, and it is deliberately **not** part of ``pepsi.target``: a site may already refresh tokens by other means. Enable it only when you want Pepsi to do the refreshing. The contract ------------ Any refresher — Pepsi's or your own — must satisfy four things: #. **Keep the client secrets out of the stage.** Put the credential-source configuration in a *separate* file and pull it into ``pepsi.conf`` with the taler ``@inline-secret@
`` directive (see :manpage:`pepsi.conf(5)`). Make that file readable only by the refresher's user. ``@inline-secret@`` is special precisely here: a reader that *cannot* open the file (the stage, running as ``pepsi``) silently skips the section and parses the rest of the configuration normally, whereas a plain ``@inline@`` would error. So the secret never reaches the stage, yet both programs share one ``pepsi.conf``. #. **Write the credential where the stage reads it, atomically.** Replace the ``TOKEN_FILE`` contents with a write-to-temp-then-``rename`` so the stage, which may read at any instant, never sees a partial file. The file content is exactly the credential the stage expects (for OAuth: the bare access token, whitespace-trimmed). #. **Get the group and mode right.** The token files live in a directory that is SGID the ``pepsi-token`` group (default ``/var/pepsi/tokens``), so a new file inherits that group; write it mode ``0640``. The consuming **stage binary is installed SGID** ``pepsi-token``, which is what lets the dispatcher's unprivileged ``pepsi`` worker read the token — without making ``pepsi`` a permanent member of the group. Any long-lived secret the refresher persists for itself (e.g. a rotated refresh token) belongs in a *private* directory (mode ``0700``), never in the group-readable token directory. #. **Fail safe.** On a refresh error, leave the previous token file in place and retry with back-off; never truncate a working token because the endpoint was briefly unreachable. The relay stage already treats a missing/expired token as a *transient* delivery failure, so the message simply waits and retries once a fresh token lands. Replacing the bundled refresher ------------------------------- To substitute your own implementation, keep the ``pepsi-helper-token-refresh`` service disabled (it is shipped disabled) and have your program write the same ``TOKEN_FILE`` with the same permissions. Nothing in the stage, ingress or dispatcher changes — the stage only ever *reads* the file. ``pepsi-setup`` prints a reminder whenever a smarthost uses ``AUTH = oauth`` so you do not forget to wire a refresher up. The same pattern generalises to any future stage that needs a periodically rotated secret: give it a ``*_FILE`` option, read the file per use, put the issuing credentials behind ``@inline-secret@``, and run the issuer as its own user writing through a group-scoped, SGID-stage-readable file. Kerberos (``AUTH = gssapi``) ---------------------------- The smarthost relay's ``AUTH = gssapi`` follows the same external-refresh contract, but the rotated credential is a Kerberos **ticket** in a credential cache rather than a token file, and the refresher is a **standard tool** rather than a Pepsi binary — so Pepsi ships none. The contract is: #. **Acquire and renew the ticket out of band.** Run ``k5start``/``krenew`` (or a ``kinit -k`` cron job) under a more-trusted user, authenticating from a *keytab* that the relay stage never reads. This obtains and periodically renews a ticket-granting ticket for Pepsi's own principal. #. **Write the cache where the stage reads it.** Point ``k5start`` at a ``FILE:`` credential cache under the SGID ``pepsi-token`` directory ``/var/pepsi/krb5`` (so the file inherits the group), and name it in the MTA's ``KRB5CCNAME`` option (or set ``KRB5CCNAME`` on the dispatcher service environment for the default cache). The SGID smarthost stage binary — already SGID ``pepsi-token`` for OAuth/mTLS — is what lets the ``pepsi`` worker read it. #. **Fail safe.** A missing or expired ticket makes the delivery a *transient* failure, so the message waits and retries once a valid ticket is in the cache. ``pepsi-setup`` prints a reminder (and sanity-checks ``FILE:`` cache readability) whenever a smarthost uses ``AUTH = gssapi``. Keeping the feature-stability table current =========================================== Every user-visible feature is listed in the single :doc:`feature-stability` table. That page is **generated** by ``contrib/update-feature-stability.sh``, which merges two sources: * the hand-maintained registry ``contrib/feature-registry.tsv`` — one row per feature, with its ``key`` (the stable telemetry identifier), human-readable name, governing RFC, automated-test coverage (``U``/``I``/``I/U``/``-``), manual-test status (``yes``/``no``) and whether it is instrumented for telemetry (``yes``/``no``); and * a ``pepsi-telemetry`` ``GET /telemetry/report`` — the per-feature deployment and usage counts, joined to the registry on ``key``. A feature whose ``telemetry`` column is ``no`` cannot be meaningfully counted. Such a row renders with a ``(*)`` after its name and ``—`` in its Deployments, Uses and Stability cells; it has **no** ``telemetry_update`` call. There are four reasons a row is in that state, and the reason must be written down beside the key in ``KNOWN_UNINSTRUMENTED`` (``tests/feature_registry.rs``) — a test checks that the list and this column agree, so that "not measured yet" can be told apart from "will never be measured": * **A passive protocol capability** the client never signals its use of (``ENHANCEDSTATUSCODES``). Note that ``PIPELINING`` is *not* one of these: a client that sends a command without waiting for the reply is directly observable, and ingress reports it once per session. * **A universal facility** whose count would measure process starts rather than use — structured logging, the telemetry channel itself, ``pepsi-setup`` (every deployment runs it, so the count would duplicate the deployment count). * **A property of a configuration or a schema** rather than an event: the ``pepsi-crypto`` grant on the private-key column, whose "use" is the *absence* of a permission error. * **A short-lived process**: an operator CLI, or a privileged ``pepsi-helper-*``. This one is a hard limit of the producer API rather than a matter of taste, so it is worth stating precisely. ``telemetry_update`` hands the feature name to a background task and returns; the task then needs a turn of the Tokio runtime to connect to the local ``pepsi-telemetry-client`` socket and write. A process that returns from its ``block_on`` immediately after reporting is dropped before that turn happens, and the event is lost — measured at **0 deliveries in 200 runs**, against 30 out of 30 when the process stays in the runtime one millisecond longer (``pepsi-common/tests/telemetry_short_lived.rs`` pins this). So the threshold is not "the process must live for a while" but "the runtime must turn over once more". A long-lived daemon or stage worker is unaffected; a CLI cannot report the thing it has just finished doing, which is the only interesting moment it has. Wiring the operator tools therefore needs a bounded, awaited ``flush()`` on the producer API first — and a per-row notion of scale, because the stability tiers are calibrated on messages (1 000 uses for ``used``, 100 000 for ``stable``) and a command a human runs a dozen times a year would be pinned ``experimental`` for ever by a counter that was working perfectly. The programs that do install the sink, and so can report, are ``pepsi-ingress``, the stage workers (via ``run_worker``), ``pepsi-dispatch``, ``pepsi-httpd``, ``pepsi-keydisc``, ``pepsi-failure-bouncer`` in its service mode, and ``pepsi-setup`` once the wizard has written a configuration whose ``SHARE_TELEMETRY`` answer it can read back. That last condition is not a technicality: reporting anything from the interview *before* the operator has answered would submit the activity of a deployment that never consented, which is what opt-in exists to prevent. Whenever you change the code or the tests, keep the table honest: #. **Adding a feature.** Add a row to ``contrib/feature-registry.tsv``. Unless the feature is genuinely un-instrumentable (set ``telemetry`` = ``no`` — reserve this for the cases above, and add the key to ``KNOWN_UNINSTRUMENTED`` in ``tests/feature_registry.rs`` *with the reason*), set ``telemetry`` = ``yes`` *and* call ``telemetry_update("")`` in a good place in the code — somewhere the feature genuinely runs, so the count reflects real usage (and the "enabled" snapshot reflects deployments that actually use it). In a stage, use ``ctx.telemetry_update("")`` at the point the stage does its characteristic work; elsewhere call ``pepsi_common::telemetry::telemetry_update("")`` directly (ingress does this for ``starttls``, ``8bitmime``, the inbound auth checks ``spf`` / ``dkim-verify`` / ``dmarc`` / ``iprev`` / ``auth-results``, and the relay client for ``dane``, ``tlsrpt`` and ``tls-identity``). The ``key`` string in the code and the registry **must match** — that is how the telemetry counts attach to the row, and the generator's ``stderr`` warning (below) catches drift. #. **Writing a test.** Update that feature's ``test`` column: ``U`` if you added a unit / in-process ``cargo test``, ``I`` for a ``tests/*.sh`` live-pipeline script or a ``pepsi-test-stages`` end-to-end test, ``I/U`` for both. #. **Manually testing.** Set the feature's ``manual`` column to ``yes``. #. **Refreshing the counts.** Run ``contrib/update-feature-stability.sh`` (it fetches ``TELEMETRY_URL`` or reads a ``TELEMETRY_REPORT_FILE``; with neither, every count is ``0`` and every feature is ``experimental``) and commit the regenerated ``docs/manual/feature-stability.rst``. The script warns on ``stderr`` about any telemetry feature with no registry row, so a stray or renamed ``key`` is caught. Do **not** edit ``docs/manual/feature-stability.rst`` by hand — it is overwritten on every regeneration. ``cargo test --test feature_registry`` enforces all of the above without needing a telemetry deployment: every reported key has a row, every row is reachable from the sources, no key is on ``KNOWN_UNINSTRUMENTED`` while the code reports it, and the list and the ``telemetry`` column say the same thing.