The inbound pipeline driver: verify → ingest → claim → handle → mark,
over a ledger resource carrying the AshHooks.InboundDelivery extension.
ingest/4 is the sync-mode entry point a consumer's controller calls with
the raw request body (captured pre-parser via the endpoint body_reader — a
router plug cannot recover pre-parse bytes). The machine's crash-safety
contract:
- the raw payload persists BEFORE any handling (audit + verification
integrity via
payload_digest); - the unique identity + storage-level uniqueness make exactly one
:createdper delivery — concurrent or sequential duplicates get:duplicate; - a claim is a WHERE-gated update (
status == :received, a re-driveable failure, or an EXPIRED lease) that bumps the monotonic fencing token — so a redelivery of a stranded row re-drives it, never no-ops (the incumbent's silent-loss window); - marks are gated on the caller's token AND an unexpired lease — a stale owner (superseded or expired) is rejected.
Returns {:ok, :created | :duplicate, delivery} — the delivery's
status carries the outcome (:processed, :failed_retryable,
:failed_permanent); handler failures are recorded in the ledger, not
raised. Verification/config failures return {:error, error} BEFORE any
ledger write (fail closed: missing secret verifies nothing; a missing raw
body never runs).
The individual machine steps (ingest_delivery/2, claim_delivery/2,
mark_processed/3, mark_failed/5, renew/3, reap/1) are public for
composition and monitoring — the async runtime (#delivery-runtime slice)
and the reaper drive the same steps.
All ledger operations run unauthorized: the signature verification IS the trust boundary for writes, and external read surfaces remain governed by the consumer's own policies (ADR-0005).
Summary
Types
The request context. :signature is the provider's signature header
value; :headers the lowercased request headers; :scope carries values
for the ledger's declared scope_identity slots; :connection is the
per-connection provider's secret source argument.
Functions
Claims a delivery: WHERE-gated on status == :received, a re-driveable
failure, or an EXPIRED lease — bumps the fencing token atomically and
takes a fresh lease. A live foreign lease (or a terminal row) returns
{:error, :lease_held}.
Drives one inbound delivery through the full sync pipeline.
Persists the ledger row (raw payload BEFORE handling) via the
no-touch unique upsert. Classifies :created by comparing the surviving
row's client-generated id against ours.
Marks a claimed delivery failed — same token/lease gate as
mark_processed/3. permanent? selects :failed_permanent over
:failed_retryable.
Marks a claimed delivery processed — gated on the caller's token under an unexpired lease.
Re-drives deliveries whose claims died with an expired lease (crash between claim and mark). Returns the number re-driven; unexpired leases and terminal rows are left alone.
Extends the caller's lease — gated on the caller's token under an unexpired lease.
Types
@type ctx() :: %{ optional(:signature) => String.t(), optional(:headers) => %{optional(String.t()) => String.t()}, optional(:method) => String.t() | nil, optional(:request_uri) => String.t() | nil, optional(:connection) => term(), optional(:scope) => %{optional(atom() | String.t()) => term()} }
The request context. :signature is the provider's signature header
value; :headers the lowercased request headers; :scope carries values
for the ledger's declared scope_identity slots; :connection is the
per-connection provider's secret source argument.
Functions
@spec claim_delivery(module(), term()) :: {:ok, non_neg_integer(), struct()} | {:error, :lease_held | term()}
Claims a delivery: WHERE-gated on status == :received, a re-driveable
failure, or an EXPIRED lease — bumps the fencing token atomically and
takes a fresh lease. A live foreign lease (or a terminal row) returns
{:error, :lease_held}.
@spec ingest(module(), atom(), binary() | nil, ctx()) :: {:ok, :created | :duplicate, struct()} | {:error, term()}
Drives one inbound delivery through the full sync pipeline.
Persists the ledger row (raw payload BEFORE handling) via the
no-touch unique upsert. Classifies :created by comparing the surviving
row's client-generated id against ours.
@spec mark_failed(module(), term(), non_neg_integer(), String.t(), boolean()) :: :ok | {:error, :stale_token | term()}
Marks a claimed delivery failed — same token/lease gate as
mark_processed/3. permanent? selects :failed_permanent over
:failed_retryable.
@spec mark_processed(module(), term(), non_neg_integer()) :: :ok | {:error, :stale_token | term()}
Marks a claimed delivery processed — gated on the caller's token under an unexpired lease.
@spec reap(module()) :: {:ok, non_neg_integer()}
Re-drives deliveries whose claims died with an expired lease (crash between claim and mark). Returns the number re-driven; unexpired leases and terminal rows are left alone.
Rows that cannot be re-driven (inbound declaration removed, provider unresolved, a raising handler) are skipped without stopping the sweep — a poison row must never starve the rows behind it (cross-vendor finding).
@spec renew(module(), term(), non_neg_integer()) :: :ok | {:error, :stale_token | term()}
Extends the caller's lease — gated on the caller's token under an unexpired lease.