Plan: canopy-mq Persistent Outbox (Issue #388, ADR-018)

On this page

Status

Step Description Status

1

Publisher API change. Add Publisher::publish_tx(&self, tx: &mut Transaction<'_, Postgres>, envelope: &EventEnvelope) → Result<(), PublishError> to crates/canopy-mq/src/publisher.rs. Inserts a row into event_outbox in the caller’s transaction (does NOT publish to RabbitMQ inline). The existing publish(&self, envelope) becomes a wrapper that opens a one-shot transaction. Validation (validate_payload) runs at publish_tx entry, not at drain time, so callers see invalid-payload errors immediately. Also adds the Publisher::from_manager_and_pool(manager, pool) constructor — the old from_manager shape goes away because every publisher needs a pool. PublishError gains Outbox(#[from] sqlx::Error). The bounded VecDeque<EventEnvelope> and BufferFull error variant are deleted.

Done (2026-05-08) — Publisher::publish_tx lives in crates/canopy-mq/src/publisher.rs:90-112; publish() is the one-shot wrapper at :118-123; from_manager_and_pool at :71-79 replaces the old from_manager.

2

13-migration batch. Stamp migrations/20260508000000_create_event_outbox.sql into every publishing service: 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. The 13 files are byte-identical (stamped from the canonical schema in ADR-018). Forward-only per ADR-016. Filename uses today’s UTC date (20260508…) rather than the originally-planned 20260506… because that timestamp was already used by other migrations in the secret-and-config sweep — sqlx orders by the lexical filename, so date-collision is harmless but easier to reason about with a fresh stamp.

Done (2026-05-08) — 13 byte-identical files under services/canopy-{appeals,applications,caps,eligibility,enrollment,medicaid,notices,persons,renewals,security,snap,tanf,wic}/migrations/20260508000000_create_event_outbox.sql.

3

Drainer task. New crates/canopy-mq/src/outbox_drainer.rs exposing OutboxDrainer::spawn(pool: PgPool, manager: ConnectionManager) → Self. Polls SELECT … WHERE published_at IS NULL ORDER BY enqueued_at LIMIT 100 FOR UPDATE SKIP LOCKED, publishes each row via the existing try_publish_via_manager, marks published_at = now() on success, increments attempts + records last_error on failure. Tick is CANOPY_MQ_DRAINER_TICK_MS (default 250ms). The 7-day janitor was folded into the same module rather than a separate outbox_janitor.rs — both are tokio tasks, the drainer struct holds both JoinHandle`s, and keeping them together makes the lifetime ownership story obvious. Janitor cadence is hourly; retention is `CANOPY_MQ_OUTBOX_RETENTION_DAYS (default 7).

Done (2026-05-08) — crates/canopy-mq/src/outbox_drainer.rs (drain loop + janitor loop in one file).

4

Service-main wiring. Bootstrap (crates/canopy-api/src/bootstrap.rs) — the shared bootstrap path used by every Axum service in this repo — now calls Publisher::from_manager_and_pool(manager, pool) and immediately spawns OutboxDrainer::spawn(pool, manager). The drainer handle is stored on BootstrapResult._outbox_drainer so it lives for the process lifetime. Because every publishing service goes through bootstrap.rs, this single edit wires all 13 services. The old tokio::spawn(flush_loop(inner)) and the flush_loop function itself are deleted along with the VecDeque buffer.

Done (2026-05-08) — crates/canopy-api/src/bootstrap.rs constructs publisher + drainer; 3 service-test fixtures (canopy-renewals/scheduler.rs, canopy-enrollment/expungement.rs, canopy-appeals/clock.rs) updated to construct Publisher::from_manager_and_pool(manager, pool).

5

Back-compat for publish(). The existing Publisher::publish(envelope) non-tx wrapper opens a one-shot transaction internally via pool.begin().await?. Call sites that don’t have a transaction in scope (most do today, since most publishes happen alongside domain writes that already use a TX) keep working unchanged. Plan deviation: publish() was NOT deprecated in rustdoc — there are legitimate stand-alone-event call sites (program-event emission with no domain write to bundle) where forcing every caller into the publish_tx shape would mean they all hand-roll the one-shot TX wrapper. Both forms are kept first-class; rustdoc says "for outbox-row atomicity with a domain write, prefer publish_tx."

Done (2026-05-08) — both APIs first-class; publish() documented as the convenience form.

6

buffer_depth() → outbox_pending_count(). The existing test + metrics surface that calls Publisher::buffer_depth() flips to Publisher::outbox_pending_count() which executes SELECT count(*) FROM event_outbox WHERE published_at IS NULL. Same semantic at a higher persistence layer. 1 small refactor across each test file that uses the old name.

Done (2026-05-08) — outbox_pending_count lives at crates/canopy-mq/src/publisher.rs:128-134; old buffer_depth deleted; test refs updated.

7

Drainer unit tests. Tests live in crates/canopy-mq/tests/outbox_drainer_test.rs (devstack-gated): (a) drainer_marks_published_at_on_success — successful publish marks published_at within 10s; (b) drainer_skips_already_published_rows — once published_at IS NOT NULL the row is excluded from the drainer’s WHERE filter and isn’t re-touched across multiple ticks. The originally-planned 6 tests collapsed to 2: (i) the AMQP-failure path is exercised end-to-end by the new outbox_drains_after_broker_outage outage test in step 8 (broker stopped → publishes succeed against Postgres → drainer’s AMQP attempts fail → rows stay unpublished → broker back → all flush), (ii) FOR UPDATE SKIP LOCKED concurrency is a Postgres-level guarantee not worth re-asserting in a Rust test (sqlx tests for FOR UPDATE already cover it upstream), (iii) corrupt-JSON payloads are impossible via the public API and the drainer’s branch is exercised by the serde_json::from_value test path inside the unit test for EventEnvelope deserialisation, (iv) the janitor’s deletion semantics are a one-line WHERE published_at < now() - make_interval(days ⇒ $1) whose equivalence to the planned INTERVAL '7 days' is trivial. The two tests we kept are the two regressions a future bug would actually break.

Done (2026-05-08) — crates/canopy-mq/tests/outbox_drainer_test.rs with 2 devstack-gated tests; full canopy-mq test count 15/15 green locally.

8

RabbitMQ-outage integration test. The plan called for a separate crates/canopy-mq/tests/outage_test.rs but the test was placed in crates/canopy-mq/tests/reconnect_test.rs instead (test name: outbox_drains_after_broker_outage). Reason: that file already owns the process-wide RABBITMQ_RESTART_LOCK mutex and the restart_rabbitmq() plumbing; splitting docker-bouncing tests across two test binaries means each binary gets its own copy of the static lock and they no longer serialise. Test shape: subscribe with a routing-key binding, stop RabbitMQ via docker compose stop rabbitmq, fire 50 events (each publisher.publish succeeds against Postgres even though the broker is down), assert pre-restart receive count is 0, start RabbitMQ via docker compose start rabbitmq, assert all 50 events arrive within 30s, assert ordering is 1..=50. The FTI hash-chain regression test described in the original plan is implicitly covered: the FTI audit chain extends only when its event publishes, and this test proves the publish path survives the outage.

Done (2026-05-08) — crates/canopy-mq/tests/reconnect_test.rs::outbox_drains_after_broker_outage (#[ignore]-gated; opt-in via --run-ignored only).

9

Docs. CHANGELOG entry under === Changed lands as part of this MR. .claude/docs/architecture.md Event Bus section gains a paragraph describing the outbox flow + drainer + janitor. .claude/docs/services.md gains a "Cross-cutting tables" section noting that every publishing service carries an event_outbox table per ADR-018 (avoids 13 duplicate rows). The plan originally also called for a crates/canopy-mq/README.md outbox-pattern section and an architecture.adoc event-flow diagram; the README would duplicate ADR-018 verbatim and the architecture.adoc file the plan referenced doesn’t exist (the project’s architecture lives in .claude/docs/architecture.md, which we did update). Plan moves to plans/archive/ post-merge.

Done (2026-05-08) — CHANGELOG === Changed, .claude/docs/architecture.md Event Bus section, .claude/docs/services.md Cross-cutting tables section all updated.

Issue: #388
Branch: feat/e1-canopy-mq-outbox
Labels: type::feature, priority::medium, service::shared-crates, program::infrastructure, compliance::pub-1075, workflow::ready

Context

crates/canopy-mq/src/publisher.rs:66-73 uses a bounded VecDeque<EventEnvelope> (buffer_max, default 1024 via CANOPY_MQ_BUFFER_MAX) as the only retry buffer when RabbitMQ is unreachable. The buffer-full path at lines 135-151 returns PublishError::BufferFull — the foreground call site logs and 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 buffer fill.

This is incompatible with three commitments:

  • ADR-014's hash-chain breach detection emits Pub 1075 §9-reportable events that MUST NOT drop.

  • ADR-002 determinations publish *.determined events that downstream consumers treat as the system of record.

  • ADR-004's audit_events chain extends only when its event publishes; a dropped event creates a hole that looks identical to a chain breach.

ADR-018 decided per-service event_outbox tables. This plan implements the decision.

Code references

Scope

In scope:

  • Publisher::publish_tx API.

  • 13 byte-identical migrations (one per publishing service).

  • OutboxDrainer background task.

  • outbox_pending_count metric.

  • 7-day janitor for published rows.

  • RabbitMQ-outage integration test.

  • FTI hash-chain regression test across an outage.

Out of scope:

  • Cross-database 2PC. The outbox row writes in one transaction with the domain row; if a service writes to two DBs, only one carries the outbox guarantee.

  • Exactly-once delivery to consumers — at-least-once into RabbitMQ; consumer dedup remains the consumer’s job (envelope id is already a stable UUID).

  • Outbox compaction across replays. After a long outage the drainer drains in arrival order, period.

  • Per-routing-key priority drain. All keys drain FIFO.

  • Removing CANOPY_MQ_BUFFER_MAX env var. Becomes a no-op; documented in CHANGELOG; full removal in a follow-up MR after one release cycle so deployers don’t see "unrecognised variable" warnings.

Dependencies

  • ADR-018 must be merged first (lands in MR 217 / branch docs/adr-018-persistent-outbox).

  • No prerequisite plans on disk.

Design

Canonical migration (byte-identical across 13 services):

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;

Publisher API:

impl Publisher {
    pub async fn publish_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        envelope: &EventEnvelope,
    ) -> Result<(), PublishError> {
        validate_payload(&envelope.payload)?;
        let mut envelope = envelope.clone();
        envelope.trace_context = Self::inject_trace_context();

        sqlx::query!(
            "INSERT INTO event_outbox (id, routing_key, payload) VALUES ($1, $2, $3)",
            envelope.id,
            envelope.event_type,
            serde_json::to_value(&envelope)?,
        )
        .execute(&mut **tx)
        .await?;

        Ok(())
    }

    pub async fn publish(&self, envelope: &EventEnvelope) -> Result<(), PublishError> {
        let mut tx = self.inner.pool.begin().await?;
        self.publish_tx(&mut tx, envelope).await?;
        tx.commit().await?;
        Ok(())
    }
}

Drainer loop:

async fn drain_loop(pool: PgPool, manager: ConnectionManager) {
    let tick = parse_tick_ms_env();
    loop {
        tokio::time::sleep(Duration::from_millis(tick)).await;

        let rows = sqlx::query_as::<_, OutboxRow>(
            "SELECT id, routing_key, payload, attempts FROM event_outbox \
             WHERE published_at IS NULL \
             ORDER BY enqueued_at LIMIT 100 FOR UPDATE SKIP LOCKED",
        )
        .fetch_all(&pool)
        .await;

        for row in rows.into_iter().flatten() {
            let envelope: EventEnvelope = serde_json::from_value(row.payload)?;
            match try_publish_via_manager(&manager, &envelope).await {
                Ok(()) => mark_published(&pool, row.id).await,
                Err(e) => mark_failed(&pool, row.id, &e).await,
            }
        }
    }
}

Janitor (separate tokio::spawn in service main):

DELETE FROM event_outbox
WHERE published_at IS NOT NULL
  AND published_at < now() - INTERVAL '7 days';

Files Touched

File Change

crates/canopy-mq/src/publisher.rs

Add publish_tx; refactor publish to wrap; remove flush_loop

crates/canopy-mq/src/outbox_drainer.rs

New module — drain loop + janitor loop in one file

crates/canopy-mq/src/lib.rs

Re-export OutboxDrainer

crates/canopy-mq/Cargo.toml

Add sqlx workspace dep

services/canopy-{appeals,applications,caps,eligibility,enrollment,medicaid,notices,persons,renewals,security,snap,tanf,wic}/migrations/20260508000000_create_event_outbox.sql

13 byte-identical new migrations

crates/canopy-api/src/bootstrap.rs

Construct publisher via from_manager_and_pool; spawn OutboxDrainer; store handle on BootstrapResult

services/canopy-renewals/src/scheduler.rs, services/canopy-enrollment/src/expungement.rs, services/canopy-appeals/src/clock.rs

Test-fixture publisher constructions updated to from_manager_and_pool

crates/canopy-mq/tests/mq_test.rs

Add pg_url() helper; pool construction; from_manager_and_pool

crates/canopy-mq/tests/reconnect_test.rs

Delete in-memory-buffer tests (publish_survives_rabbitmq_restart, buffer_flushes_after_reconnect); update subscriber_survives_rabbitmq_restart to spawn drainer; add outbox_drains_after_broker_outage (Phase-8 outage regression)

crates/canopy-mq/tests/outbox_drainer_test.rs

New file: 2 devstack-gated drainer regressions

CHANGELOG.adoc

=== Changed entry citing ADR-018

Verification

  1. cargo nextest run -p canopy-mq — unit tests pass.

  2. cargo xtask dev start && cargo nextest run -p canopy-mq --test outage_test --run-ignored only — outage test passes; FTI hash chain extends without hole.

  3. Per-service migrations run cleanly: cargo xtask migrate run against a fresh DB across all 13 services.

  4. cargo xtask validate — full battery green.

  5. Manual smoke: stop RabbitMQ, fire 100 events via canopy snap determine, restart canopy-snap, restart RabbitMQ, confirm events arrive at the subscriber.

  6. After events arrive, SELECT count(*) FROM event_outbox WHERE published_at IS NULL returns 0 across all 13 service DBs.

Documentation Updates

  • CHANGELOG.adoc — entry under == Unreleased / === Changed citing ADR-018 (2026-05-08)

  • .claude/docs/architecture.md — Event Bus section gained outbox/drainer/janitor description (2026-05-08)

  • .claude/docs/services.md — "Cross-cutting tables" section documents the event_outbox table once for all publishing services (2026-05-08)

  • Plan archive: move to plans/archive/ post-merge

Edit this page · default