18. 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-<name>] 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 <ingress_id> 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).

18.1. 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 pauses, fails 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.

18.2. 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:

1. Library crate pepsi-stage-<name> 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).

2. Binary src/bin/pepsi-stage-<name>.rs that calls taler_main, exposes a worker subcommand delegating to the library’s worker, and runs the library’s run for a bare <message_id> argument.

3. Register the crate in the workspace Cargo.toml and add the binary to the Makefile’s BINARIES list.

18.3. 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-<name>] 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.<key>). 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("<key>") — 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).

18.4. Terminal helpers (choose exactly one)

The terminal helper you call is the contract with the dispatcher:

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.

18.5. 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).

18.6. 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.

18.7. 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-<name> <id> -c test.conf) — the global -c flag must follow the message-id/subcommand.

18.8. Adding support for migrating from another MTA

pepsi-setup can import an existing mail server’s configuration as the wizard’s answer defaults (see Installation and 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<Detected>;
    fn import(&self, probe: &Probe, found: &Detected) -> Imported;
}

Four rules make the difference between an importer that helps and one that misleads:

1. 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.

2. 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).

3. 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/<mta>/ with no access to the real system.

4. 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.

18.8.1. Testing an importer

Every fixture tree lives at pepsi-setup/tests/fixtures/<mta>/<case>/ and is a probe root: the directory stands in for /, so pepsi-setup import <mta> --root <that directory> 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_<mta>.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.

18.9. 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 (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 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.

18.9.1. 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.

18.9.2. The contract

Any refresher — Pepsi’s or your own — must satisfy four things:

1. 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@ <SECTION> <FILE> directive (see 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.

2. 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).

3. 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.

4. 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.

18.9.3. 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.

18.9.4. 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:

1. 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.

2. 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.

3. 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.

18.10. Keeping the feature-stability table current

Every user-visible feature is listed in the single 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:

1. 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("<key>") 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("<key>") at the point the stage does its characteristic work; elsewhere call pepsi_common::telemetry::telemetry_update("<key>") 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.

2. 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.

3. Manually testing. Set the feature’s manual column to yes.

4. 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.