Plan: canopy-api hardening + canopy-mq consumer inbox (Issues #437 #433)

On this page

Status

Step Description Status

1

crates/canopy-api/src/lib.rs: replace /healthz (line 134, handler at 269-329) with /livez (200 always, process-alive only; no DB/MQ checks) + /readyz (current health_check behavior; 503 if DB or MQ degraded). Add HSTS header (Strict-Transport-Security: max-age=31536000; includeSubDomains; preload) to the SetResponseHeaderLayer stack at lines 146-156. No back-compat alias for /healthz — pre-1.0.

Not started

2

crates/canopy-api/src/admin.rs (new): AdminRoutes builder helper that mounts /v1/admin/* routes behind a service-class JWT + actor.has_role("admin") gate. Single endpoint today: POST /v1/admin/events/replay accepting {event_ids: [Uuid]} JSON body, returning canopy_mq::ReplayReport. Public API: AdminRoutes::router().with_replay(pool, mq_conn).build() returns an axum Router services merge into their main router.

Not started

3

crates/canopy-mq/src/inbox.rs (new): event_inbox schema constants + insert/select helpers. Schema: event_id UUID PRIMARY KEY, routing_key TEXT NOT NULL, payload JSONB NOT NULL, enqueued_at TIMESTAMPTZ NOT NULL DEFAULT now(), processed_at TIMESTAMPTZ, attempts INT NOT NULL DEFAULT 0, last_error TEXT. PK is event_id (NOT auto-gen) — matches EventEnvelope.id (UUID v7 per envelope.rs:22). Helpers: try_insert(&mut tx, envelope) → Result<InsertOutcome> where InsertOutcome is Inserted / AlreadyProcessed / InFlightRetry; mark_processed(&mut tx, event_id); bump_attempts(pool, event_id, err: &str) → Result<i32> (returns new count, NON-TX so it survives handler-rollback); find_by_ids(pool, ids: &[Uuid]) → Result<Vec<InboxRow>>.

Not started

4

crates/canopy-mq/src/subscriber.rs: breaking API change. Today there are 4 public variants (subscribe, subscribe_with_options, subscribe_with_dlx, subscribe_exclusive) all delegating to subscribe_inner. Rewrite: (a) DELETE subscribe_with_dlx — DLX is auto-derived from queue name in all variants (exchange: "canopy.dlq", queue: format!("{queue_name}.dlq"), routing_key: queue_name). (b) The remaining 3 variants — subscribe, subscribe_with_options, subscribe_exclusive — gain pool: PgPool + max_attempts: u32 parameters and a new handler signature Fn(EventEnvelope, &mut Transaction<', Postgres>) → Fut<Result<(), anyhow::Error>>. Return type stays Result<JoinHandle<()>, lapin::Error>. (c) Per-message flow in subscribe_inner: BEGIN TX → inbox::try_insert(&mut tx, envelope) ON CONFLICT DO NOTHING → if conflict and processed_at IS NOT NULL (already handled): commit + ack + skip handler → if conflict and processed_at IS NULL (in-flight retry): proceed → call handler(envelope, &mut tx) → on Ok: inbox::mark_processed(&mut tx, id) + commit + ack → on Err: rollback + inbox::bump_attempts(&pool, id, &err) (non-TX so the counter survives rollback) → if attempts < max_attempts: nack(requeue=true) else nack(requeue=false, routes to DLQ). HRTB lifetime on the handler signature: resolve with for<'a> Fn(EventEnvelope, &'a mut Transaction<', Postgres>) → BoxFuture<'a, Result<(), anyhow::Error>> or equivalent.

Not started

5

crates/canopy-mq/src/replay.rs (new): pub async fn replay_messages(pool: PgPool, manager: ConnectionManager, event_ids: &[Uuid]) → Result<ReplayReport>. Reads event_inbox rows where event_id = ANY($1). Classifies each requested id: row with processed_at IS NOT NULL → already_processed; row with processed_at IS NULL → eligible; not in inbox → missing. Reconstructs EventEnvelope from inbox row + re-publishes via Publisher::publish(&envelope) (one-shot, not via outbox — operator-initiated). Returns ReplayReport { replayed: Vec<Uuid>, missing: Vec<Uuid>, already_processed: Vec<Uuid> }.

Not started

6

crates/canopy-mq/src/lib.rs: re-export inbox module + replay_messages + ReplayReport. Drop DlxConfig from public surface (becomes internal-only after subscribe_with_dlx removal).

Not started

7

13 service migrations + janitor task. Migrations: services/canopy-{enrollment,renewals,medicaid,notices,security,snap,tanf,wic,caps,eligibility,appeals,applications,persons}/migrations/20260516000000_create_event_inbox.sql. Schema mirrors the ADR-018 event_outbox per-service pattern (reference: services/canopy-medicaid/migrations/20260508000000_create_event_outbox.sql). Partial index on (enqueued_at) WHERE processed_at IS NULL for the replay hot path. Forward-only per ADR-016. Plus a new janitor task in crates/canopy-mq/src/inbox_drainer.rs mirroring OutboxDrainer: 24h sweep deleting event_inbox rows where processed_at < now() - 7 days. Spawned in BootstrapResult alongside _outbox_drainer.

Not started

8

7 subscriber call-site updates to new handler signature. Files: services/canopy-{enrollment,medicaid,notices,snap,tanf,web,security}/src/main.rs. Each handler accepts (envelope, &mut tx); domain work uses the supplied TX so inbox-insert + domain-write commit atomically. subscribe_with_dlx(…​) call in services/canopy-medicaid/src/main.rs:166 → replaced with new subscribe(…​) shape (DLX auto-derived from "canopy-medicaid.tma" queue name). canopy-security’s wildcard # subscriber stays on subscribe_exclusive (new shape).

Not started

9

Admin replay endpoint wired per subscribing service. Each of the 7 subscriber services adds the admin route at boot: router.merge(AdminRoutes::router().with_replay(boot.db.inner().clone(), boot.mq_connection.clone()).build()). Each endpoint reads from its own service’s event_inbox and calls replay_messages.

Not started

10

Dockerfile HEALTHCHECK + xtask + e2e updates: every HEALTHCHECK CMD curl …​ /healthz → /livez. Find via grep -rn "/healthz" across services/canopy-*/Dockerfile, xtask/src/, tests/e2e/.

Not started

11

Tests. canopy-api: 4 new unit tests (livez_returns_200_even_when_db_unreachable, readyz_returns_503_when_mq_degraded, hsts_header_present, admin_replay_rejects_non_service_caller_and_non_admin_actor). canopy-mq: 8 new tests (inbox_try_insert_is_idempotent, subscriber_dedupes_redelivery_via_inbox, subscriber_commits_on_handler_ok, subscriber_rolls_back_on_handler_err, inbox_bump_attempts_increments, subscriber_max_attempts_routes_to_dlq, replay_messages_re_publishes_from_inbox, replay_report_classifies_correctly). 7 subscriber service integration-test handler signature updates.

Not started

12

Plan filed (this file). CHANGELOG entry under === Changed with explicit breaking-change callouts (/healthz removed, subscribe API replaced). docs/modules/ROOT/pages/adrs/adr-018-persistent-outbox.adoc gets an amendment paragraph documenting the consumer-side inbox extension. docs/modules/ROOT/pages/data-models/canopy-<service>.adoc × 13 updated for event_inbox. Service Catalog subscribing-service sections updated.

Not started

13

Precommit Q1-Q8 answered via subagent verification per .githooks/pre-commit rule from ae8251f. Substantial diff (~2000 LOC) — verifier scrutiny warranted.

Not started

Issues: #437, #433
Branch: feat/canopy-api-mq-hardening
Labels: priority::medium, service::shared-crates, program::infrastructure, type::feature, compliance::pub-1075, workflow::ready

Context

Two coupled production-hardening gaps surfaced in the 2026-05-09 external review, both port-backs from CRAIG:

#437 (canopy-api): the current shared bootstrap collapses liveness and readiness into a single /healthz endpoint that checks BOTH the process is up AND deps (DB, RabbitMQ) are reachable. Kubernetes / k8s-shaped orchestrators behave better when these are separated — liveness failures restart the pod, readiness failures pull from service-mesh traffic. The bootstrap also lacks HSTS (no Strict-Transport-Security in the security-headers stack) and has no /admin/* route family for operator surgery actions.

#433 (canopy-mq): canopy has the producer half of the durable-events story (event_outbox per ADR-018) — a domain write + outbox insert commit in one TX, then a drainer publishes to RabbitMQ. The consumer half is missing. When a subscriber’s handler fails after partial domain work, the message either redelivers (and double-applies) or dead-letters and the human work is lost. CRAIG has the matching pattern: per-subscriber event_inbox table, idempotent handler invocation, retry/backoff up to a max, then DLQ, with an admin endpoint to replay DLQ’d events after operator triage.

The two ship together because the admin replay endpoint surface (POST /v1/admin/events/replay) lives in canopy-api’s new admin family (#437) but the replay logic reads from canopy-mq’s inbox (#433). Each is incomplete without the other.

User direction (2026-05-14): no back-compat

Pre-1.0; no shims, aliases, or deprecation periods. /healthz is removed (not aliased). Subscriber::subscribe is replaced wholesale with the new inbox-aware variant. Subscriber::subscribe_with_dlx is removed entirely (DLX auto-derived). All Dockerfile HEALTHCHECK lines update to /livez in this MR. All 7 subscriber call sites update to the new handler signature in this MR.

Code references

  • crates/canopy-api/src/lib.rs:134, 269-329 — current /healthz route + handler.

  • crates/canopy-api/src/lib.rs:146-156 — SetResponseHeaderLayer stack (HSTS added here).

  • crates/canopy-api/src/bootstrap.rs:17-50 — BootstrapResult shape; carries db: DbPool, mq_connection: ConnectionManager.

  • crates/canopy-mq/src/subscriber.rs:68-152 — current 4 public subscribe variants.

  • crates/canopy-mq/src/subscriber.rs:345-408 — current ack/nack loop (rewritten with inbox).

  • crates/canopy-mq/src/envelope.rs:18-57 — EventEnvelope.id: EventEnvelopeId (UUID v7).

  • crates/canopy-mq/src/publisher.rs:118 — Publisher::publish(&envelope) for replay re-publish.

  • crates/canopy-mq/src/outbox_drainer.rs:59-82 — OutboxDrainer::spawn pattern to mirror for the inbox janitor.

  • services/canopy-medicaid/src/main.rs:150-166 — only existing subscribe_with_dlx callsite.

  • crates/canopy-auth/src/claims.rs:134, 227, 254 — has_role, require_service_caller, actor helpers.

  • services/canopy-medicaid/migrations/20260508000000_create_event_outbox.sql — reference shape for event_inbox migration.

Scope

In scope (single MR feat/canopy-api-mq-hardening):

  • All 13 Status-table steps land together. ~2000 LOC code + ~600 LOC tests.

  • Breaking changes: /healthz removed; 3 subscribe variants gain inbox semantics; subscribe_with_dlx deleted.

  • 13 event_inbox table migrations (matches event_outbox per-service footprint).

  • 7 subscriber service handler migrations.

  • Per-service POST /v1/admin/events/replay endpoint, gated by service-class JWT + actor.has_role("admin").

  • Dockerfile HEALTHCHECK updates to /livez.

  • xtask + e2e references to /healthz updated.

Out of scope:

  • True exponential backoff via delayed retry queue. Needs RabbitMQ delayed-message plugin or per-queue TTL hops — separate infra concern. Max-attempts-then-DLQ is the substitute.

  • Auto-replay on schedule. The admin endpoint is operator-initiated. Scheduled replay-on-cooldown is post-UAT.

  • Cross-service admin endpoint that walks every inbox. Per ADR-001, each service owns its own inbox.

  • Removing MqHealth from /readyz. Stays — readiness includes RabbitMQ.

  • OpenAPI documentation of /livez + /readyz. Operational endpoints, not API surface.

  • Migrating non-subscribing services to use the new subscriber API. They have no consumers — they get the event_inbox table prophylactically (consistency with event_outbox).

Dependencies

  • Independent of other Tier 1 issues (#438 ✅, #435 ✅, #436).

  • Convention deps: ADR-013, ADR-016, ADR-018 (extends), ADR-001.

  • No new workspace deps (sqlx, tokio, tracing, lapin, serde_json, uuid all already pulled in).

Design

/livez and /readyz

// crates/canopy-api/src/lib.rs

.route("/livez", get(livez_check))    // 200 always while process runs
.route("/readyz", get(readyz_check))  // 503 if DB or MQ degraded
// /healthz REMOVED — no back-compat alias.

async fn livez_check() -> impl IntoResponse {
    (StatusCode::OK, "ok")
}

// readyz_check = the existing health_check at :269-329 with status-code
// flipped to 503 on degraded.

HSTS header layer after the existing 3 security headers (:146-156):

.layer(SetResponseHeaderLayer::overriding(
    http::header::STRICT_TRANSPORT_SECURITY,
    http::HeaderValue::from_static("max-age=31536000; includeSubDomains; preload"),
))

AdminRoutes builder

// crates/canopy-api/src/admin.rs (new file)

pub struct AdminRoutes;

impl AdminRoutes {
    pub fn router() -> AdminRoutesBuilder { AdminRoutesBuilder::default() }
}

#[derive(Default)]
pub struct AdminRoutesBuilder {
    replay: Option<(PgPool, ConnectionManager)>,
}

impl AdminRoutesBuilder {
    pub fn with_replay(mut self, pool: PgPool, manager: ConnectionManager) -> Self {
        self.replay = Some((pool, manager));
        self
    }

    pub fn build(self) -> Router {
        let mut r = Router::new();
        if let Some((pool, manager)) = self.replay {
            r = r.route(
                "/v1/admin/events/replay",
                post(admin_replay_handler).with_state(AdminReplayState { pool, manager }),
            );
        }
        r
    }
}

async fn admin_replay_handler(
    Extension(claims): Extension<canopy_auth::Claims>,
    State(state): State<AdminReplayState>,
    Json(req): Json<ReplayRequest>,
) -> Result<Json<canopy_mq::ReplayReport>, ApiError> {
    claims.require_service_caller()?;
    let actor = claims.actor().ok_or(ApiError::Forbidden)?;
    if !actor.has_role("admin") { return Err(ApiError::Forbidden); }
    let report = canopy_mq::replay_messages(
        state.pool.clone(), state.manager.clone(), &req.event_ids,
    ).await.map_err(|e| ApiError::internal("replay failed", e))?;
    Ok(Json(report))
}

event_inbox migration (byte-identical across 13 services)

-- services/canopy-<service>/migrations/20260516000000_create_event_inbox.sql

CREATE TABLE event_inbox (
    event_id       UUID PRIMARY KEY,
    routing_key    TEXT NOT NULL,
    payload        JSONB NOT NULL,
    enqueued_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    processed_at   TIMESTAMPTZ,
    attempts       INT NOT NULL DEFAULT 0,
    last_error     TEXT
);

CREATE INDEX event_inbox_unprocessed_idx ON event_inbox (enqueued_at)
    WHERE processed_at IS NULL;

COMMENT ON TABLE event_inbox IS
    'Per-service consumer inbox (#433). Subscriber INSERT ON CONFLICT DO NOTHING on each delivery — handler invocation is idempotent across redelivery. Same pattern as event_outbox (ADR-018) but on the consumer side.';

Per-message flow (Step 4 in detail)

For each delivery:

  1. Deserialize EventEnvelope. On error → nack(no-requeue) → DLQ.

  2. let mut tx = pool.begin().await?.

  3. let outcome = inbox::try_insert(&mut tx, &envelope).await?:

    • Inserted → proceed to (4)

    • AlreadyProcessed → commit + ack + skip handler

    • InFlightRetry (row exists, processed_at IS NULL) → proceed to (4)

  4. Call handler(envelope, &mut tx).await.

  5. On Ok(()): inbox::mark_processed(&mut tx, id) + tx.commit() + delivery.ack().

  6. On Err(e): tx.rollback() + inbox::bump_attempts(&pool, id, &e.to_string()) (non-TX) → if attempts < max → nack(requeue=true) else nack(requeue=false → DLQ).

Dockerfile HEALTHCHECK migration

HEALTHCHECK CMD curl -f http://localhost:PORT/healthz || exit 1 → /livez. Liveness is the right probe for Docker; readiness is operator-monitor concern.

Files Touched

File Change

crates/canopy-api/src/lib.rs

Replace /healthz with /livez + /readyz. Add HSTS layer.

crates/canopy-api/src/admin.rs

New: AdminRoutes builder + admin replay handler.

crates/canopy-mq/src/inbox.rs

New: inbox helpers (try_insert, mark_processed, bump_attempts, find_by_ids).

crates/canopy-mq/src/subscriber.rs

Delete subscribe_with_dlx. Rewrite the 3 remaining variants with inbox semantics + handler taking &mut Transaction.

crates/canopy-mq/src/replay.rs

New: replay_messages + ReplayReport.

crates/canopy-mq/src/inbox_drainer.rs

New: janitor task (7-day cleanup). Spawned in BootstrapResult.

crates/canopy-mq/src/lib.rs

Re-export inbox, replay_messages, ReplayReport. Drop DlxConfig from public surface.

services/canopy-<service>/migrations/20260516000000_create_event_inbox.sql × 13

New migrations (byte-identical schema).

services/canopy-{enrollment,medicaid,notices,snap,tanf,web,security}/src/main.rs

7 subscriber call-site updates: new handler signature + auto-DLX queue name + admin route merged.

services/canopy-*/Dockerfile

HEALTHCHECK CMD curl …​ /healthz → /livez.

xtask/src/ + tests/e2e/

Grep /healthz and update to /livez where appropriate.

docs/modules/ROOT/pages/adrs/adr-018-persistent-outbox.adoc

Amendment paragraph: consumer-side inbox extension.

docs/modules/ROOT/pages/data-models/canopy-<service>.adoc × 13

Add event_inbox to Tables list + ERD + Migration-files list.

CHANGELOG.adoc

Entry under == Unreleased / === Changed. Breaking-change callouts.

Service Catalog

Subscribing-service sections updated.

OpenAPI snapshot regeneration: required (canopy-api gains /v1/admin/events/replay). Run cargo xtask api-docs --update.

Verification

  1. cargo nextest run -p canopy-mq -p canopy-api --lib — new + existing tests pass.

  2. cargo nextest run --workspace — all subscriber service integration tests pass with new handler signature.

  3. cargo fmt --check --all + cargo clippy --all-targets — -D warnings — zero warnings.

  4. cargo xtask api-docs --update — new admin endpoint reflected in snapshots; verify diff.

  5. cargo xtask validate — full battery green.

  6. Manual smoke: cargo xtask dev refresh. curl localhost:8000/livez → 200. curl /readyz → 200. Stop RabbitMQ; /readyz → 503; /livez → still 200.

  7. Manual smoke (replay): publish an event the consumer rejects; after max retries it lands in DLQ; POST /v1/admin/events/replay {event_ids:[…​]} with service-class+admin JWT — confirm ReplayReport.replayed includes the id.

  8. Adversarial smoke: replay without admin role → 403; without service-class → 401/403.

Documentation Updates

  • Plan filed at docs/modules/ROOT/pages/plans/canopy-api-mq-hardening.adoc.

  • CHANGELOG.adoc — entry with explicit breaking-change callouts.

  • docs/modules/ROOT/pages/adrs/adr-018-persistent-outbox.adoc — amendment.

  • 13 service data-model pages updated.

  • Service Catalog — subscribing-service updates.

  • Plan moves to plans/archive/ post-merge per ADR-013.

Why this approach (vs alternatives)

  • Don’t keep /healthz as back-compat alias. User explicitly chose no shims pre-1.0.

  • Don’t add the new subscriber API alongside the old. Half-migration leaves long-tail debt.

  • Don’t make DLX configurable per-subscriber. Auto-derive removes boilerplate; canopy-medicaid (the only DLX user today) already follows the convention.

  • Don’t add cross-service admin endpoint. ADR-001 — each service owns its own inbox.

  • Don’t implement true exponential backoff in this MR. Needs RabbitMQ delayed-message plugin; separate infra concern.

  • Don’t make handler signature take &Pool instead of &mut Transaction. Atomic inbox-insert + domain-write requires TX.

Risk + Rollback

  • Risk: external monitoring or k8s manifests probing /healthz will break. Mitigation: CHANGELOG entry calls out the breaking change explicitly.

  • Risk: 7 subscriber handler updates may have TX-semantic regressions. Mitigation: per-handler integration tests must pass; reviewer scrutinises each diff.

  • Risk: inbox INSERT-ON-CONFLICT adds DB load proportional to message volume. Mitigation: indexed PK; partial index on WHERE processed_at IS NULL.

  • Risk: replay re-publishes raw envelope payload — if consumer logic has changed since original publish, replay produces different results. Mitigation: intended semantic (replay-after-bugfix); operator decision.

  • Rollback: revert the MR. /healthz returns. Subscribe API reverts. event_inbox migrations stay (ADR-016 forward-only) but unused. No data loss.

Branch + label hygiene

  • Branch: feat/canopy-api-mq-hardening

  • Type: type::feature (matches both issues)

  • Priority: priority::medium (matches both)

  • Service: service::shared-crates

  • Program: program::infrastructure

  • Compliance: compliance::pub-1075 (production-hardening ATO posture)

  • Workflow: workflow::ready → workflow::in-progress at branch → workflow::in-review at MR open

Edit this page · default