ADR-018: Persistent Per-Service Event Outbox

On this page

Context

crates/canopy-mq/src/publisher.rs:66-73 uses a bounded VecDeque<EventEnvelope> (default cap 1024 envelopes via CANOPY_MQ_BUFFER_MAX) as the only retry buffer when RabbitMQ is unreachable. Once the buffer fills, enqueue returns PublishError::BufferFull and the foreground call site logs + drops the event. On process restart the buffer is gone too — any envelopes that arrived between the broker outage and the restart are lost, regardless of how full the buffer was.

This is incompatible with several existing commitments:

  • ADR-014's hash-chain breach detection emits fti.audit_chain.breach_detected events that are Pub 1075 §9-reportable. A dropped breach event silently weakens the audit posture.

  • ADR-002 determinations publish *.determined events that downstream consumers (canopy-enrollment for SNAP issuance, canopy-notices for NOA generation, canopy-reporting for federal reports) treat as the system of record. A dropped determination event leaves the orchestrator’s DB in programs_approved while the downstream systems never see it — silent inconsistency.

  • ADR-004's audit_events chain extends only when the corresponding event publishes successfully. A dropped audit event creates a chain hole that the chain-status endpoint cannot distinguish from an integrity breach.

The transactional-outbox pattern (write the event row in the same DB transaction that writes the domain row, drain to the broker asynchronously) is the standard solution. The decision space is where the outbox lives.

Three options were considered:

  1. Per-service event_outbox table. Each publishing service owns its own table in its own DB. Publisher API gains publish_tx(&mut tx, …) that writes the outbox row in the caller’s transaction. A background drainer (started in service main()) selects unpublished rows and pushes to RabbitMQ.

  2. Shared canopy-outbox service. A dedicated outbox microservice with its own DB; every publisher posts via HTTP. Cleaner abstraction but introduces a single point of failure that contradicts ADR-001 program isolation, and adds an HTTP hop in the publish path.

  3. Embedded in canopy-mq with pluggable backend. OutboxStore trait in canopy-mq with a Postgres implementation; each service injects its own pool. Lighter shared-code footprint than option 1 but each service still owns its table.

Decision

Per-service event_outbox table. Each publishing service owns its own table in its own database. Publisher writes the outbox row in the caller’s transaction; a background drainer in the same process flushes to RabbitMQ.

This preserves ADR-001 (no cross-service DB writes), aligns with the existing per-service migration model (ADR-016), and keeps the publish path in-process (no extra hop, no new SPOF).

The publisher API gains:

impl Publisher {
    /// Persists the envelope into the caller's transaction and returns. The
    /// background drainer publishes to RabbitMQ later. The contract is:
    /// once `publish_tx` returns Ok and the caller commits, the event WILL
    /// reach the broker eventually (modulo bug-for-bug-equivalent failures
    /// where the database itself is unrecoverable).
    pub async fn publish_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        envelope: &EventEnvelope,
    ) -> Result<(), PublishError>;
}

The existing publish(&self, envelope) becomes a wrapper that opens a one-shot transaction; call sites that already have a transaction in scope migrate to publish_tx. The bounded VecDeque and flush_loop are removed; their job is now done by the drainer reading from Postgres.

Canonical schema (lands as migrations/20260506000000_create_event_outbox.sql in every publishing service):

CREATE TABLE event_outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    routing_key TEXT NOT NULL,
    payload JSONB NOT NULL,
    enqueued_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at TIMESTAMPTZ,
    attempts INT NOT NULL DEFAULT 0,
    last_error TEXT
);

CREATE INDEX event_outbox_unpublished_idx
    ON event_outbox (enqueued_at)
    WHERE published_at IS NULL;

Filename + body are byte-identical across the 13 publishing services (canopy-snap, canopy-tanf, canopy-medicaid, canopy-caps, canopy-wic, canopy-applications, canopy-eligibility, canopy-enrollment, canopy-renewals, canopy-appeals, canopy-notices, canopy-security, canopy-persons). sqlx migrations are per-service in each service’s migrations/ directory; there is no shared-template mechanism in sqlx, so the file is hand-stamped from this ADR. New publishing services inherit the same stamp.

Drainer policy:

  • Polls every 250ms when the in-process buffer was previously busy; backs off to 1s when the broker has been steady for a minute. Tunable via CANOPY_MQ_DRAINER_TICK_MS (matches the existing CANOPY_MQ_* env-var convention).

  • Selects up to 100 unpublished rows per tick (SELECT … WHERE published_at IS NULL ORDER BY enqueued_at LIMIT 100 FOR UPDATE SKIP LOCKED), publishes serially in enqueued_at order, marks published_at = now() on success.

  • On AMQP failure: increments attempts, records last_error, leaves published_at null. The next tick retries. Reconnect is driven inline via the existing ConnectionManager::reconnect single-flight (same primitive flush_loop used to call).

  • FOR UPDATE SKIP LOCKED lets multiple replicas of the same service drain concurrently without double-publishing. Existing single-replica deployments behave identically.

  • Rows older than 7 days with published_at IS NOT NULL are deleted by an in-process janitor running once per hour. Tunable via CANOPY_MQ_OUTBOX_RETENTION_DAYS.

Out of scope:

  • Cross-database transactional outbox — when a publisher writes to two databases (rare; the only current case is some bootstrap admin scripts), the outbox-row write happens in one of the two transactions, not both. The other write must be idempotent or eventually-consistent. No cross-DB 2PC.

  • Exactly-once delivery to consumers. The outbox guarantees at-least-once publish to RabbitMQ; consumer-side de-duplication remains the consumer’s responsibility (envelope id is already a UUID and stable across retries).

  • Outbox compaction across replays. After a long outage, the drainer floods the broker in enqueued_at order. Rate-limiting + flow control are RabbitMQ’s job, not the outbox’s.

  • Per-routing-key priority. All routing keys drain in arrival order. If a future audit-event ordering requirement needs priority drain, a follow-up issue will revisit.

Consequences

Positive

  • No event loss across broker outages or process restarts. Outbox rows survive restart; drainer picks up on next start.

  • FTI hash chain (ADR-014) and audit_events chain (ADR-004) extend without holes — chain-extension events publish in the same TX as the chained row, so chain integrity tracks domain integrity.

  • No new external infrastructure. Each service already has a Postgres database; the table sits alongside existing domain tables.

  • Trivially reviewable migrations. 13 byte-identical SQL files; review is a checksum exercise.

  • Per-service isolation preserved (ADR-001). No shared outbox service, no cross-DB writes.

  • Multi-replica safe. FOR UPDATE SKIP LOCKED lets the drainer scale horizontally without coordination.

Negative

  • Each publish costs one extra Postgres write. For services that publish 10s-100s of events per request (e.g., audit_events extension on every FTI access), this is a measurable per-request cost. Mitigation: publish_tx shares the caller’s existing transaction, so it’s one extra row per existing TX, not a new round-trip.

  • Drainer adds a long-running task per service. One more thing to watch in service-main wiring; mitigated by spawning at the same place Publisher::from_manager already runs (just replace the body of the existing flush_loop spawn).

  • Outbox table grows unbounded if the drainer breaks. The 7-day janitor only deletes published rows. Stuck rows accumulate. Mitigation: an alert on event_outbox row count > N (tunable; default 10k) emits to canopy-security via the existing audit_events log path. Operator response is the same as for any drainer-stuck condition: investigate why publishes are failing, fix, drain catches up. (Delivered 2026-07-29 as #1230, scale audit L2 — with two deltas from this sketch: the alert surface is the drainer’s own stats task — canopy_mq_outbox{pending,oldest_unpublished_age_seconds,parked} gauges, a WARN log, and the /readyz outbox check — not an audit_events write; and an oldest-unpublished-age bound (default 900s) alarms alongside the count, because a small wedged backlog never tops 10K. Poison rows additionally PARK after CANOPY_MQ_DRAINER_MAX_ATTEMPTS row-culpable failures (default 10) and are replayable via the #433 admin surface, which unparks and resets the budget.)_

  • Migrations land in 13 services in one MR. Large blast radius; mitigated by the bytes-identical property (one diff to read, applied 13 times) and by the fact that migrations are forward-only (ADR-016) — if one service’s migration trips, that service is the only one stuck while others advance.

Neutral

  • The bounded VecDeque + flush_loop go away. Existing CANOPY_MQ_BUFFER_MAX env var becomes a no-op; documented in CHANGELOG and removed in a follow-up MR after one release cycle so deployers don’t see "unrecognised variable" warnings during the transition.

  • Publisher::buffer_depth() (used by tests + /healthz metrics in some services) becomes Publisher::outbox_pending_count() (counts unpublished rows). Tests update; metrics endpoints get a free upgrade — depth is now a stable durable quantity, not a transient in-memory one.

  • This ADR amends the implicit contract behind ADR-014's breach_detected event delivery: previously "best-effort with bounded retry"; now "at-least-once to RabbitMQ with durable storage in front."

Amendment (2026-05-14) — Consumer-side inbox (#433)

ADR-018 originally specified producer-side outbox: domain write + outbox INSERT commit in one TX, drainer flushes to RabbitMQ. Issue #433 extends the same pattern to the consumer side. Every subscriber service now owns a per-database event_inbox table (byte-identical schema across the 13 service migrations under 20260516000000_create_event_inbox.sql); the Subscriber writes a row in the handler’s transaction before invoking the handler. PK on event_id (the envelope’s UUID v7) makes redelivery idempotent — a duplicate envelope hits ON CONFLICT DO NOTHING and the subscriber acks without re-running the handler.

Per-delivery flow on the consumer side:

  1. BEGIN TX

  2. inbox::try_insert → InsertOutcome::{Inserted, InFlightRetry, AlreadyProcessed}

  3. If AlreadyProcessed: commit, ack, skip handler.

  4. Else: invoke handler with &mut tx. On Ok: mark_processed + commit + ack. On Err: rollback + bump_attempts (separate connection so the counter survives rollback) + nack(requeue=true) until attempts >= max_attempts, then nack(no-requeue) → DLQ.

The DLX is auto-derived (canopy.dlq exchange, <queue>.dlq queue, queue name as routing key). The Subscriber::subscribe_with_dlx variant from #417 is removed in the same MR — DLX wiring is no longer a per-caller concern.

The producer + consumer halves together close the at-least-once delivery loop: events emitted by a domain transaction reach the consumer’s domain transaction with exactly-once-effect, gated by the inbox PK and the subscriber’s transactional commit. Replay is operator-initiated via POST /v1/admin/events/replay on the subscribing service.

The 7-day janitor pattern extends: InboxDrainer::spawn(pool) deletes processed rows older than CANOPY_MQ_INBOX_RETENTION_DAYS (default 7 days). Unprocessed rows (processed_at IS NULL) are preserved indefinitely so operator-initiated replay can find them.

Amendment (2026-05-18) — Lease-based drainer (#478)

The original OutboxDrainer::drain_once implementation opened a single Postgres transaction that spanned the batch’s N RabbitMQ publishes. Under workspace integration load this serialised foreground COMMITs behind the WAL writer: the drainer’s per-row UPDATE event_outbox SET published_at = now() WHERE id = $1 accumulated WAL records that the next foreground COMMIT had to fsync past, producing repeated LWLock:WALWrite + IO:WalSync waits with COMMITs measuring multiple seconds. pg_stat_activity polling captured the antipattern as drainer sessions sitting idle in transaction on the per-row UPDATE while domain sessions waited on COMMIT.

This amendment establishes one hard invariant: no broker I/O inside a database transaction. The drainer now operates in three phases — claim, publish (no tx), mark — using two new lease columns on event_outbox:

  • claimed_at TIMESTAMPTZ NULL — when the current claim began.

  • claimed_by TEXT NULL — the drainer identity that holds the claim, sourced from CANOPY_MQ_REPLICA_ID or falling back to drainer-{hostname}-{pid}-{uuid-v7}.

Plus a new partial index event_outbox_lease_idx ON event_outbox (claimed_at NULLS FIRST, enqueued_at) WHERE published_at IS NULL that keeps the Phase 1 claim query cheap. The existing event_outbox_unpublished_idx is retained during transition; redundancy is assessed in a follow-up after observed query plans.

Phase 1 is a single CTE statement using the canonical skip-locked-claim pattern (FOR UPDATE SKIP LOCKED inside the CTE; outer UPDATE stamps the lease and returns). Phase 2 acquires one channel per batch, calls confirm_select once, then pipelines publishes up to CANOPY_MQ_DRAINER_PIPELINE_DEPTH deep before awaiting confirms. mandatory is deliberately not set — in a fan-out topic exchange "no queue currently bound" is a normal condition, and mandatory=true would treat such events as failed and retry them indefinitely. Phase 3 marks results in two short transactions: a bulk UPDATE … SET published_at = now() for confirmed rows and a bulk UPDATE … SET attempts = attempts + 1, last_error = … for failed rows, both guarded by WHERE id = ANY($1) AND claimed_by = $drainer_id. The claimed_by guard means a slow drainer whose lease has already expired silently no-ops on rows another drainer has reclaimed.

Crashes mid-batch are recovered by the next drainer tick: the Phase 1 query reclaims rows whose claimed_at < now() - CANOPY_MQ_DRAINER_LEASE_TTL_SECS. Lease recovery is the claim path’s job, not the hourly janitor’s (which continues to handle only retention sweep of published rows). attempts is not incremented at claim time — only on per-message publish failure — so reclaimed rows from a crashed drainer don’t inflate the retry counter.

The producer-side guarantee from the original ADR is unchanged: an event is durably written iff the domain transaction commits. The at-least-once delivery contract is also unchanged — subscribers must remain idempotent via the event_inbox ON CONFLICT pattern from the consumer-side amendment above (issues #437 / #433). A publish that succeeds at the broker but whose Phase 3 mark fails (process crash, lease expiry mid-confirm) re-publishes on a subsequent tick; subscribers absorb the duplicate at their event_inbox row.

Configuration adds three env vars in the existing CANOPY_MQ_* style:

  • CANOPY_MQ_DRAINER_BATCH_SIZE (default 100)

  • CANOPY_MQ_DRAINER_LEASE_TTL_SECS (default 60)

  • CANOPY_MQ_DRAINER_PIPELINE_DEPTH (default 32)

DrainerConfig::from_env asserts at boot that batch_size > 0, pipeline_depth > 0, and lease_ttl_secs >= 3 × pipeline_depth × 100ms (the assumed worst-case per-publish latency) so misconfiguration fails fast rather than silently at the first tick.

See outbox-drainer-lease-refactor for design rationale and the four in-source verification tests.

Amendment (2026-07-16) — Bounded Phase-2 publish+confirm wait (#1061)

The lease-based drainer’s Phase 2 awaited broker interactions — channel open, confirm_select, each publish, each publisher confirm — with no deadline. A broker that stopped answering confirms (or black-holed TCP that stalled any of those awaits) parked the entire claimed batch invisibly: the tick never returned, the rows stayed claimed until lease expiry, and only another drainer’s reclaim recovered them. Found and verified during the #1059 flake diagnosis (not causal there).

Phase 2 now runs under one deadline, CANOPY_MQ_DRAINER_CONFIRM_TIMEOUT_SECS (default 30), covering the whole publish+confirm exchange for a batch. On expiry the batch takes the existing infra-error path: every claim is released without bumping attempts (the rows are blameless), the tick returns a typed ConfirmTimeout error, and the drain loop drives the single-flight broker reconnect from the #1060 fix (a wedged confirm implicates the connection the same way an AMQP error does). Rows whose confirms had already landed in the interrupted batch republish on a later tick; the at-least-once contract is unchanged and the consumer-side event_inbox absorbs the duplicates.

Boot asserts extend accordingly: confirm_timeout_secs × 1000 >= pipeline_depth × 100ms (one worst-case batch, so a merely-slow broker is not misread as wedged) and confirm_timeout_secs < lease_ttl_secs (a timed-out batch must release its claims before another drainer reclaims them via lease expiry).

Implementation tracker

Implementation lands under plan plans/canopy-mq-persistent-outbox.adoc (Issue #388). The 13 service migrations land in the same MR as the publisher refactor; the drainer wiring lands in each service’s main.rs as part of that MR. The consumer-side inbox extension lands under plan plans/canopy-api-mq-hardening.adoc (Issues #437 + #433); 13 byte-identical event_inbox migrations + 7 subscriber call-site rewrites + the new /v1/admin/events/replay admin endpoint family ship together.

References

Edit this page · default