Runtime and delivery

Every stage of a pipeline runs as its own task, connected to its neighbors by bounded channels of Arrow record batches.

Backpressure is structural

A channel holds at most a configured number of batches. When the sink is slower than the source, the channel fills and the source blocks on send — no unbounded buffering, no load shedding, no special-case code. Memory in flight is bounded by channel capacity × batch size, derived from runtime.maxInflightBytes in the pipeline document.

This is tested behaviorally: with a deliberately blocked sink, the source is verified to stall after filling the channels.

Failure and cancellation

• First failure wins. A failing stage cancels a shared token before its channels close, so downstream stages can always tell “upstream finished” from “upstream failed”. The user sees the root-cause error with its stage id, never the teardown noise.
• Ctrl-C is cooperative. External cancellation uses the same token; every potentially blocking await in every stage races against it, so shutdown is prompt even mid-batch.

Typed faults

Every induced failure surfaces as a documented, typed error — never a panic, a hang, or an untyped string — and a fault-injection suite pins this. Stage failures are classified as invalid data (user-fixable: schema, value, configuration) or external system (object store, database, provider). Provider calls carry a finer taxonomy:

Fault	Meaning	Typical response
timeout	The request exceeded its deadline (2 min default for openai-compat, tunable)	Retry; environmental
throttled	Rate or quota rejection (HTTP 429 and equivalents)	Back off and retry
transport	Connection refused, reset, DNS failure	Retry; check the endpoint
protocol	The provider answered outside its protocol shape	Check endpoint/model configuration
server	Provider-side failure (HTTP 5xx)	Provider incident; wait or escalate

Retrying is always safe bookkeeping: completed work is in the ledger and is never re-dispatched, and the sink’s transactionality means a failed run committed nothing. The suite also kills the database backend mid-COPY and verifies the run fails typed with the target table untouched.

Commit safety

The sink finalizes its work (transaction commit, staging-table merge) only from an explicit commit step that runs after the input stream ended cleanly and nothing upstream failed:

• run succeeds → one commit, rows become visible atomically;
• any stage fails → no commit, the target table is untouched;
• cancelled mid-run → no commit.

Checkpoints and incremental runs

With runtime.checkpoint configured, the source is enumerated into work units — one immutable file each, identified by path + size + modification time. Before reading, completed units are skipped and pending ones are durably claimed; after the sink commits, the consumed units are durably marked complete. The store is an append-only, fsync’d JSONL log that survives kill -9 mid-append (a torn final record is detected and discarded; corruption anywhere else fails loudly instead of losing data).

The practical effect: re-running a finished pipeline does nothing, and a run over a grown directory loads only the new files. This works the same for local directories and s3:// prefixes — S3 unit identity (key, size, last-modified) comes from a single LIST request per run.

runtime:
  checkpoint:
    url: file:///var/lib/pramen/checkpoints/

Delivery contract

The v1 contract is at-least-once (ADR 0006): a run that fails after partial progress is safe to re-run — nothing was committed, nothing was checkpointed. The one duplication window is a crash between the sink’s commit and the checkpoint completion record; with mode: append those units’ rows would repeat on the next run. mode: upsert closes that window: rows are staged in a session-local temporary table and merged into the target with INSERT … ON CONFLICT on the declared key columns, so replays update in place instead of duplicating (the target needs a unique index over exactly those columns). Both sides of the contract are pinned by tests: a replayed append run demonstrably duplicates, the same replay under upsert demonstrably does not. Semantic (ai.*) work is exactly-once-billed regardless: duplicated loads reuse ledger results. Exactly-once delivery from input positions alone is never claimed.

sink:
  type: postgres
  target: analytics.events
  mode: upsert
  keys: [id]