ADR-003: RabbitMQ Event Topology

On this page

Status

Accepted

Context

CRAIG services need to communicate asynchronously for domain events (e.g., a case being created triggers placement evaluation, rules evaluation triggers case updates). We needed to design a messaging topology that supports both competing consumers (for DB-mutating handlers) and fan-out (for cache invalidation).

Decision

Use a single topic exchange (craig.events) with service-owned queues and dotted routing keys.

Exchange

  • Name: craig.events

  • Type: topic

  • Durable: true

Routing Keys

Events use the format {service}.{event_type}:

  • rules.evaluations_archived (#1129 / ADR-058: rules' fleet-bookkeeping fact — one archive batch of rule_evaluations archived-and-pruned; staged inside the prune transaction. Payload = the batch manifest facts incl. the D9 store identity + archived_by; craig-security validate-then-writes it into archive_records (D16 — envelope fields are data, not transport identity; forged/malformed envelopes are refused loudly and land only in the generic audit row). Rules' own archive_ledger — not this event — is the authority its orphan GC consults)

  • security.archive_completed (REDEFINED by #1129: now marks a REAL archive act — one batch of security’s own audit-class rows archived-and-pruned, payload = manifest facts + archived_by; the pre-#1129 event announced a stale purge_eligible sweep), security.archive_purged (#1129 D15: one expired archive destroyed; carries purged_by)

  • rules.evaluated, rules.cache_invalidated, rules.authz_fallback_admit (rules.evaluated is a pointer payload since #1130 — rule-set identity + evaluation_id + context, never the input/output documents; the single durable copy is the producer’s rule_evaluations row; since #1556 it also carries rule_set_revision, the content-exact identity joining straight to rule_set_snapshots. #786: authz_fallback_admit staged on every craig-rules bootstrap fallback admit; attribution payload, no PII)

  • rules.ruleset_promoted, rules.ruleset_rolled_back (#1556 / D2: the operator lifecycle flips of live rule policy, staged on the flip transaction — rule_set_id, name, version, from_revision→to_revision, user_id (the acting operator, feeding the security actor chain), reason, for promote the preview_id pointing at the persisted canary evidence, and for rollback the restored_revision (the snapshot the content came FROM — from_revision is the live state it moved off). Security parses them as promote/rollback on rule_set with the rule_set_id resolved — never the raw-split fallback)

  • authz.access_denied (ADR-050 §Amendment #1063 — staged by the craig-authz engine on every EXPLICIT PolicyDenied row-check outcome, volume-gated by the full-dimension per-emitter aggregate: first occurrence per (sub, resource_type, resource_id, action, ruleset_name, denied_reason) per 60 s window emits kind: "deny" (acting worker under sub, user_roles, resource + action + jurisdiction, ruleset_name/ruleset_version, categorical denied_reason, suppressed_prior when a prior window closed, aggregate_overflow at cap); drained windows emit kind: "suppression-flush" with the suppressed count on the SAME routing key — consumed by craig-security’s wildcard audit subscriber like the sibling authz events)

  • authz.cache_miss, authz.cache_refreshed (ADR-050 / #908 — staged by the craig-authz engine inside EVERY hosting service via its injected OutboxAuditSink; source_service names the emitter. Miss payload: ruleset/jurisdiction/resource-type lookup + acting-worker attribution (sub pseudonymous UUID, is_service, service_id — no PII), emitted ONLY on source-attested absence, never on PolicyLoad faults. Refresh payload: trigger rmq_event|ttl_refresh|boot|warmup|lazy_load + version diff, new_version: null = upstream eviction)

  • case.referral_created, case.intake_created, case.created, case.updated, case.closed

  • placement.created, placement.activated, placement.ended, placement.requested (placement.activated is the billing trigger — its payload is the shared typed craig-placement-contracts DTO both the producer and the craig-financial consumer compile against; #979), placement.home_document_uploaded, placement.home_document_upload_failed (#1251 / ADR-062 §U — staged in the same transaction as the finalize/reap; the failed payload keys on attempt_id and carries the sanitized file_name so reap-path failures name the file, no other PII)

  • exchange.sent, exchange.failed, exchange.received, icpc.created, icpc.status_changed

  • case.screening_cohort_changed, exchange.screening_run_invalidated (#1467 / ADR-065 D1 rule 2 + ADR-066 — the SSA-screening pair, catalogued + parsed BEFORE any producer exists per D17/H25, the ADR-057 runbook precedent; B6b ships the emitters. screening_cohort_changed is the fork-5 MINIMAL shape — case_id + the monotonic cohort_revision + a categorical change token ∈ {household, person_digest, ssn_cleared, link_promoted}; deliberately NO person_id and NO SSN digest, person-level tokens are activity-revealing to every queue reader — consumed by craig-exchange’s inbox (watermark GREATEST-upsert → strict < invalidation → sweep) and audited as update/screening_cohort so the derived signal never double-counts as a case update. screening_run_invalidated carries run_id + the denormalized case_id + reason ∈ {cohort_changed, operator_cancelled, freshness_expired}, plus present-only superseded_by_run_id (the fork-3 auto-supersede) and present-only user_id = the relayed requested_by sub on operator_cancelled ONLY (the fork-1 actor-naming obligation; system variants carry no attribution key). NO sweep counts ride either payload — the DB stamps are the accounting (the ADR-066 as-built posture); audited as update/ssa_screening_run with run_id resolved before case_id (the adjustment-family shape))

  • exchange.screening_hold_set, exchange.screening_hold_cleared (#1566 / UD10 — the legal-hold transition pair, staged in the SAME transaction as the run-row flag flip; one event per genuine TRANSITION (a replay of an already-held/already-clear run stages nothing, so the audit trail IS the transition history). Payload: run_id + the denormalized case_id, user_id = the acting admin’s sub (a hold is never a system act — the human-admin-only endpoint gate is the fork-6 screening_resume posture), and the operator-entered bounded reason (the hold’s narrative lives HERE, never on the run row — the flag is a bare boolean, the archive_records precedent). Audited as update/ssa_screening_run with run_id resolved via the exchange.screening_ family; same same-MR parser-arm deploy-order note as ssa.jobs_resumed)

  • ssa.jobs_resumed (#1469 / ADR-066 fork 6 — ONE event per operator resume act, staged in the same tx as the fork-6 attempts-reset: user_id = the resuming admin’s sub, the filters as sent (present-only reason_filter/run_id), and each re-admitted job with its PRE-reset attempts count. Batch grain — no singular resource id (the payments_voided shape); audited update/ssa_screening_job. Categorical only: ids, counts, closed tokens. Deploy-order note: the parser arm ships in the same MR as the producer — safe because the wildcard subscriber’s raw-split fallback records unknown types losslessly; a security instance older than the arm labels rows (jobs_resumed, ssa) until restarted, never drops them)

  • case.screening_review_dismissed (#1557 / UD5: ONE event per human dismissal of a screening-review worklist item — the audited decision that new SSA data does NOT warrant re-determination, staged in the dismiss transaction. Payload: item_id, case_id (the audited subject; resolves via the generic selector), user_id (the acting worker), the categorical latest_reason the item carried, and the bounded free-text dismissal_reason (the substance of the decision — no PII beyond what the worker types; the UI labels the field audit-visible). Parsed as update/screening_review)

  • ssa.stale_cohorts_invalidated (#1489 / ADR-065 fork 4 — ONE summary per executed rotation sweep, staged BEFORE the per-run stamps as the sweep INTENT: user_id = the sweeping admin’s sub, current_version, and run_ids + count = the CANDIDATE set the operator executed against. Every stamp that follows is a subset of the declared candidates, so a crash mid-sweep can never leave an invalidation without an attributing summary; each stamped run carries its own exchange.screening_run_invalidated system-variant event (reason cohort_changed) as the completion evidence, and a post-crash re-run declares a fresh intent over the remainder. Batch grain, categorical only; audited update/ssa_screening_run via the explicit parser arm (same fallback-safe same-MR deploy-order note as ssa.jobs_resumed))

  • financial.payment_created, financial.payment_approved, financial.payment_issued, financial.payment_cleared, financial.payments_voided, financial.claim_submitted, financial.claim_accepted (#1079 — federal acceptance recorded on a claiming record; ids/period/actor only), financial.adjustment_created, financial.adjustment_resolved (#1078 — the payment-correction pair: ids/actor only, resolution = approved | denied; no reason text, no money), financial.subsidy_generation_completed (payments_voided is one bulk event per set-based void — voided ids + count + cause (placement_ended | subsidy_reconciliation, #1068) with present-only scope keys (placement_id | agreement_id + month), no PII; #953. The three lifecycle events share one payload shape carrying the acting worker’s sub as user_id; #978 made issue/clear reachable and stopped approval re-emitting payment_created. payment_created + the lifecycle events OMIT case_id when the payment has none — post-case subsidy rows, #1068. subsidy_generation_completed is one event per generation run: run id, month, initiator kind, and the report counters — no PII; ADR-053)

  • financial.subsidy_agreement_imported, financial.subsidy_import_batch_finalized (#1071 / ADR-057 — the conversion import; the PRODUCER landed with the MR-B surface, the parser deploy-first: one imported event per agreement MATERIALIZED at batch finalize (agreement/child/program/head-status/slot-count/batch/cutover — the source-system record reference deliberately does NOT ride events; it is readable via the records ledger / detail API / web provenance block by authorized operators), and one counters-only summary per finalized batch. No PII in either payload)

  • reporting.quality_issue_detected, reporting.afcars_transmitted, reporting.ncands_transmitted, reporting.afcars_export_generated, reporting.ncands_export_generated, reporting.afcars_export_failed, reporting.ncands_export_failed (#1252 / ADR-062 §U — the export-attempt pointer flips and reconciler reaps; generated payloads carry the artifact digest + generation, failed payloads key on attempt_id and carry the deterministic filename, no PII)

  • security.review_due, security.archive_completed, security.major_change_recorded, security.major_change_reported, security.alert_created, security.alert_acknowledged

  • case.report_submitted, case.report_disposition_recorded, case.report_attachment_uploaded, case.report_converted (cases owns the report lifecycle per ADR-017)

  • case.plan_approved, case.plan_superseded, case.plan_updated, case.task_created, case.task_updated, case.task_deleted, case.person_created, case.person_updated (cases owns the case-plan / task / person mutation lifecycle; 821 — all carried to the audit subscriber; person payloads carry no PII), case.person_pii_exported (#1064 / ADR-051 — the federal-export READ-audit: requestor service, person-id list, include_full_ssn, purpose; the repo’s first read-access audit event, still zero PII), case.person_ssn_admin_action (#1064 — SSN verification revoked / SSN cleared, carrying the worker-entered reason to audit_log; the one person payload with worker-authored text, deliberately kept OUT of tracing logs)

  • security.partner.created, security.partner.key_issued, security.partner.key_revoked, security.partner.signer_key_registered, security.partner.signer_key_approved, security.partner.signer_key_revoked (security owns partner identity + signer keys per ADR-018)

Wildcard Audit Subscriber

craig-security binds the queue craig-security.events to craig.events with routing key # — every event published to the exchange is written to the audit_log table by a single subscriber. This is the mechanism through which CRAIG achieves audit coverage for all domain mutations without per-service wiring. See Architecture — Cross-cutting Service Contracts for the wider pattern.

Consumer Patterns

(Retitled from "Queue Types" by the #1198 amendment below — "queue type" now unambiguously means the RabbitMQ classic/quorum/stream property, not the consumer pattern.)

  • Competing consumer queues (e.g., craig-cases.events): Shared across all instances of a service. Each message is delivered to exactly one consumer. Used for DB-mutating event handlers.

  • Exclusive queues (e.g., craig-rules.cache.{uuid}): Per-instance, auto-delete. Every instance receives every message. Used for cache invalidation.

Rationale

  • Single exchange: Simplifies topology. Topic routing provides flexible filtering without creating per-service exchanges.

  • Service-owned queues, operator-owned exchanges: Each service declares its own queues and bindings on startup. The exchanges (craig.events, craig.dlx) are provisioned by the operator (definitions.json / management API), NOT declared by services — see Amendment #1202.

  • Competing vs exclusive: This dual-queue pattern supports horizontal scaling. DB mutations must be processed once (competing); cache invalidation must reach all instances (exclusive).

  • Why not direct exchange: Topic exchange supports wildcard routing (e.g., case.*) which may be useful for future audit/logging consumers.

  • Why not pub/sub (fanout): Fanout doesn’t support selective subscription. Topic exchange provides both fan-out (exclusive queues with # binding) and point-to-point (competing queues with specific bindings).

Consequences

  • Services must handle idempotent event processing — messages may be redelivered on connection failure.

  • Event consumers are best-effort — publishers do not wait for consumer acknowledgment.

  • Adding a new event type requires no infrastructure changes — just publish with a new routing key and bind consumers.

Amendment — #1198 (2026-07-29): queue-type & delivery-limit contract (epic &75 C4)

Nothing previously pinned the RabbitMQ queue type (classic vs quorum vs stream). The fleet’s redelivery semantics — the client 2-strike nack ladder, the #1181 x-death token grammar (ADR-022), and the planned #1197 parking tier — all assume classic queues with no broker-autonomous redelivery bound, which was default-behavior-in-absence-of-config, not an invariant: RabbitMQ 4.x quorum queues default delivery-limit = 20 (breach → drop, or DLX if configured — the DLQ deliberately has none, so a quorum DLQ would silently drop forensic records), and a one-dropdown vhost default_queue_type changes what type-omitting declarations get.

Decision

  • Every CRAIG-declared queue is classic, with an explicit x-queue-type=classic argument — the production choke-point (SubscriptionSpec::prepare_queue via queue_args, all three subscribe variants) and the in-tree test helpers (EventCollector, DlxCollector). The complete argument table per role is unit-pinned by full-table equality, so x-delivery-limit (or any other arg) can never appear silently.

  • Delivery limits: a quorum-queue feature classic queues do not implement — type enforcement transitively neutralizes delivery-limit policies. The machine-checked invariant is the queue type; the no-limit posture follows from it. Residually an operator convention (devstack ships no policies; the deployment guide prohibits delivery-limit or queue-type-affecting policies/operator-policies). It is NOT runtime-enforced — services hold AMQP credentials only, no management API — and this ADR does not claim otherwise.

  • Quorum migration is a named future gate: only with broker clustering, and only through a new ADR that revisits together the #1181 x-death count grammar, at-least-once dead-lettering, and the delivery-limit posture.

What the explicit argument actually buys

RabbitMQ 4.x resolves and compares the queue type on every declare even when the client omits it, so the redeclare-conflict 406 largely pre-exists. Explicit classic adds: (1) immunity to a vhost/node default_queue_type at fresh declare — the real silent-loss hole; (2) independence from broker-version-dependent type-injection behavior; (3) auditable intent in the stored queue arguments.

Enforcement layers

  • A — declare-time: the eager first session means a conflicting pre-existing queue fails the service boot with 406 PRECONDITION_FAILED naming x-queue-type.

  • A′ — runtime: the reconnect path classifies AMQP 406 into an error!-level "operator action required" log (generic about queue-vs-exchange — exchanges are declared before queues and 406 is not type-specific) while preserving retry-forever (#1127). Degraded mode is deliberate: the service stays up and loud; readyz reflects the publisher path only.

  • B — source lint: the blocking cargo xtask lints mq-topology gate — a crate-boundary convention gate (raw declaration tokens only inside crates/craig-mq/src/, crates/craig-test-lib/src/, or test paths). It does not prove declarations carry the type; queue_args + its unit pins do, inside the boundary.

  • C — devstack pins: rabbitmq.conf sets default_queue_type = classic (node) and quorum_queue.property_equivalence.relaxed_checks_on_redeclaration = false (load-bearing: relaxing it makes the broker ignore x-queue-type on redeclare — a subscriber would silently consume from a quorum queue); definitions.json pins the / vhost metadata DQT (canonical exported nested-metadata shape). Definitions import never overwrites existing broker state — the devstack broker is volume-less, so drift is cured by container recreation; the live remedy is rabbitmqctl update_vhost_metadata / --default-queue-type classic.

Rollout & residuals

  • 4.x-born queues store the injected classic type, so the explicit-classic redeclare is equivalent — no queue wipe, no operator action on sanctioned brokers (live-pinned for both durable arg shapes). Residual: queues first declared on pre-3.13 brokers and upgraded in place may 406 at boot; the deployment guide carries the symptom and the shovel/drain remedy.

  • The exclusive fan-out variant’s explicit type is hygiene, not enforcement (uuid-named, live collision is 405, quorum structurally impossible for exclusive queues).

  • Crash-path residual — RETIRED by #1203 (2026-07-30): a handler panic no longer unwinds the supervisor. Every user-handler path (events, DLQ, inbox) runs inside a per-delivery task boundary that converts UNWIND panics to the normal failure ladder (2-strike nack / TransientPark capture / inbox error_count stamp) — the dlq_handler_panics_total{path} counter is the code-defect tell. Unwind-only: panic = "abort" builds are not contained (the workspace is unwind-everywhere).

  • Broker permission separation (per-service credentials) landed as Amendment #1202 (below) — the single administrator credential this amendment inherited is retired there.

  • Live pins: crates/craig-mq/tests/queue_contract.rs (DQT-vhost, node-default, both conflict tripwires, both rollout shapes, mid-life drift loudness).

Amendment — #1197 (2026-07-29): parking + quarantine companion queues (epic &75 C3)

ADR-059 adds two durable classic queues per DLQ subscription, declared through the same craig-mq choke-point the #1198 amendment pins (classic-only argument tables; the mq-topology lint boundary applies): {queue}.parking — a plain holding queue consumed by the client-driven replayer (prefetch 1; NO TTL or DLX arguments, so no redeclare hazard ever) — and {queue}.quarantine — terminal, no consumer, operator-drained per the deployment guide’s manual-ack replay procedure. Both are published to via the DEFAULT exchange with publisher confirms + mandatory + persistent delivery; no broker-managed dead-letter hop exists anywhere in the retry path (classic-queue dead-lettering is best-effort — rejected as a return leg by ADR-059’s review). Production footprint: craig-security.dlq.parking / craig-security.dlq.quarantine. The deployment guide’s policy MUST-NOTs are extended by #1197 (message-ttl, max-length, max-length-bytes, expires — which would DELETE the consumer-less quarantine queue — and destructive overflow on craig* queues).

Amendment — #1202 (2026-07-30): operator-owned exchanges + per-service credentials (epic &75 addendum)

Two coupled changes retire the shared single-credential posture this ADR shipped with.

Operator-owned exchange topology

Services no longer declare exchanges. craig_mq::declare_dlx and the open_channel exchange declares are removed (pre-1.0, no shims); craig.events and craig.dlx are provisioned by the operator (devstack/rabbitmq/definitions.json; production per the deployment guide). Reason: a service that declares an exchange needs configure on it, and configure includes DELETE — a compromised service could delete/redeclare craig.dlx and silently suppress forensic delivery without ever touching the DLQ, falsifying ADR-022's #1181 bound.

Recovery-semantics change (deliberate): a WIPED broker is no longer silently self-healed. Boot fails loudly; the mid-life session rebuild retries the bind under the #1127 supervisor backoff (promoting that 404-at-rebuild to the operator-action-required error class is #1210). The old silent redeclare masked the data loss a wipe implies.

Per-service least-privilege credentials

Ten accounts replace the one shared craig administrator: craig keeps administrator as the operator/management-plane identity (no service connects as it), craig-test is the broad AMQP test identity (no tags), and eight per-service accounts are each scoped by anchored regex (RabbitMQ patterns are substring matches — anchors mandatory) to their own queues plus the shared exchanges they use. The exact triples are devstack/rabbitmq/definitions.json (the executable spec) and the deployment guide re-renders them with the rabbitmqctl forms + the ordered cutover/rollback runbook.

Accepted bounds (named, not silently accepted):

  • craig-security’s `amq.default write is write-anywhere-via-default-exchange for the audit principal — per-queue default-exchange scoping is inexpressible; a dedicated capture exchange is future work.

  • Injection residuals: any service may publish any routing key to craig.events (event forgery) and any dlq.* key to craig.dlx (forensic-record forgery). Topic permissions are the future hardening; the residual is recorded now.

  • Events-plane eavesdropping: any service may bind any routing key on craig.events.

  • Enforcement line: read on craig.dlx appears only in `craig-security’s set — "only craig-security reads `dlq.*`"; delete-on-shared-exchanges is denied fleet-wide.

Devstack honesty posture: devstack credentials are a FUNCTIONAL ACL FIXTURE, not a containment boundary — the port binds the host, passwords equal the account name and are public, craig/craig-test retain broad rights on the operator/test plane. Production gets generated secrets, network restriction, and no craig-test analogue. The devstack broker is rabbitmq:4.2-management-alpine; the 4.3.1+ passive-declare ACL is not yet enforced, so the grants are forward-compatible, never relied on as current enforcement.

Live pins: crates/craig-mq/tests/broker_permissions.rs (the mgmt-exact fleet pin against definitions.json + the least-privilege probes, incl. both exchange_delete 403s) and the empty-vhost bind-refusal pin in queue_contract.rs.

Amendment — #1186 (2026-07-31): readiness scope superseded in part (epic &76)

The recurring statement "`readyz` reflects the publisher path only" (Enforcement layers A′ above, echoed in the craig-mq module doc and shared-crates) is superseded in part by ADR-061, whose epic-&76 implementation CHANGES /readyz to gate on the publisher connection (the pre-#1186 Extension wiring never delivered it — readiness has been DB-only) AND on registered critical-worker liveness, so a subscriber SUPERVISOR’s death will gate readiness once the program’s U2 lands (the ADR-061 plan’s Status table is authoritative for what has shipped). Unchanged either way: the #1127 self-heal semantics inside a live supervisor (reconnect-forever, degraded-and-loud) — a reconnecting subscriber does NOT flip readiness; only the supervisor task dying does.

Amendment — #1528 (2026-08-20): AMQP establishment carries a liveness bound

"Retry-forever" (Enforcement layer A′) presupposes the retry loop keeps ticking. It did not. lapin is constructed with ConnectionProperties::default(), and no connect, handshake, or per-RPC timeout exists anywhere in the client. The AMQP URL is not an escape hatch either: amq-protocol-uri parses a ?connection_timeout= query parameter, but the async TCP path never reads it. Nor does the heartbeat cover the gap — lapin arms the heartbeat timer only on StartHeartbeat, i.e. after Tune, so before the handshake completes no timer exists at all (afterwards there is a ~4×(heartbeat/2) killswitch, far above any liveness budget). A broker path that accepts TCP and never answers (a half-open NAT/LB entry, a paused or wedged node, a saturated proxy hop) therefore parks the awaiting task indefinitely. The #1141 shield in reconnect() raced only the shutdown token, and a running service never fires it, so ONE black-holed rebuild pinned the #1127 supervisor permanently: the capped-backoff ladder never got another tick, subscriber reconnected never logged, and the queue silently accumulated behind a service that stayed up and — per A′ — stayed "ready".

Every AMQP operation that would otherwise park a long-lived task now runs under the crate-level AMQP_LIVENESS_TIMEOUT (15 s) via one bounded_amqp helper — one bound, applied at every such site rather than a fix at the one that was observed. (Every site today; nothing structurally prevents a future unbounded Connection::connect, so this is review-enforced, not compiler-enforced.)

Bounded site Task it would otherwise pin

SubscriptionSpec::build_session (whole build, not just the dial)

the #1127 consume supervisor + both eager boot builds (supervise_regular, subscribe_dlq)

build_replayer_session

the ADR-059 §D4 DLQ parking replayer

ensure_sampler_conn + create_channel + sample_queue (all three steps of a cycle)

the #1199 DLQ depth sampler (a parked tick loop freezes the gauges at their last value, and a stale depth gauge reads as a healthy queue). The sampler REUSES its connection across ticks, so its realistic black hole is a mid-stream one — the channel open is as exposed as the dial

craig_mq::connect

every service’s boot publisher connect — now a loud failure instead of a startup that hangs with no error and no readiness

The bound wraps the WHOLE establishment, not just Connection::connect: once the TCP handshake completes, a black hole that starts mid-stream parks queue_declare / queue_bind / basic_consume identically, and there is no per-RPC timeout either.

Two contracts hold the amendment together. First, expiry maps to a TimedOut I/O error, which lapin::Error::is_io_error classifies TRANSIENT — so log_rebuild_failure keeps its warn! arm and the ladder retries; the operator-action error! arm stays reserved for the 406/404 that retrying can never clear (layer A′, #1210). That chain is driven end-to-end rather than inspected — an_expired_establish_lets_the_backoff_ladder_tick_again runs the real reconnect() against a black-holed listener and asserts from the supervisor’s own narration that the delay ADVANCES past the expiry and that no operator-action arm was taken. Second, 15 s sits between the crate’s single-RPC bounds (SETTLE_TIMEOUT 5 s, CONFIRM_TIMEOUT 10 s — unit-pinned as a floor, since a session build is several RPCs) and the contested suite’s 90 s reconnect-barrier floor (craig_test_lib::fault::mq_sever::RECONNECT_DEADLINE), at no more than a third of the latter: one stalled establish costs at most a third of a recovery envelope instead of all of it. That is an ordering pin and not a budget — it does NOT say how many attempts fit, and the arithmetic that would is wrong, because it omits the backoff: past the ladder’s 30 s cap, one stalled attempt plus one cap-length wait is 45 s, so two fill the barrier exactly. Sizing the barrier against a contested ladder remains the harness’s job (#1509); the bound’s job is that a stall costs one tick rather than every tick. The ordering is itself pinned (transport_faults.rs, no broker required) — neither crate’s own unit tests can hold an invariant that spans both. The second-order effect is accepted deliberately: an establish that legitimately needs more than 15 s now fails and restarts the ladder, i.e. a retry inside the same recovery envelope rather than a park outside every envelope.

Deliberately NOT bounded: the consume loop’s wait for the next delivery (idling on an empty queue is correct, not a stall). #1531 closed the residual settle/publish gaps: EVERY settle — the inline ack, all five nacks, the post-capture acks — is bounded by SETTLE_TIMEOUT (a hang maps to the transient TimedOut class, ends the session as SettlementFailed, and the supervisor rebuilds instead of the delivery loop parking behind an accept-and-stop-answering broker path); every basic_publish SEND (both Publisher paths and the capture/replayer primitive) is bounded by AMQP_LIVENESS_TIMEOUT, with the recorded classification: an expired send stays RETRYABLE (PublishError::Amqp carrying TimedOut; the capture arm’s Failed) — the broker may have accepted the frame, and the confirm layer plus the consumer-side inbox dedup own the resulting duplicate.

The cost: since #1530, zero stranded threads. lapin spawns its lapin-io-loop OS thread before the handshake completes and exposes no cancel-safe teardown; under the original #1528 timeout-and-drop posture each abandoned attempt stranded one thread on the black-holed socket until the peer closed (measured: exactly +1 per attempt, reaped ≤3 s after peer close). connect_owned removed the strand outright: craig-mq dials the TCP socket itself (bounded — no lapin machinery exists yet), hands the connected stream to Connection::connector through a single-shot cell, and holds a dup’d fd in a SocketReaper across the handshake — when the bound expires, or ANY outer bound drops the future mid-handshake, the reaper’s Drop shuts the socket down and the io-loop reads an error and exits immediately. owned_connect_strands_no_io_loop_threads is the regression contract on BOTH kill paths (own expiry; outer drop) — it counts lapin-io-loop threads by name and requires ZERO after each abandoned-attempt volley while the peer STILL holds the socket. The AMQPS scheme falls back to lapin’s own dial under the same bound (no deployment terminates TLS at the client today) and keeps the old strand-until-peer-close cost — recorded, not hidden. The contract reads /proc/self/task and is therefore cfg(target_os = "linux"): it holds on CI and on Linux workstations, and is simply absent elsewhere — the cost claim is unverified on a non-Linux dev box, not asserted there.

Pre-1.0 behavioral change: against an unreachable-but-accepting broker, a service boot now fails after 15 s where it previously hung indefinitely.

Edit this page · latest