Broker Phases 6–10: clustering, durability, transactions, auth, management - #18
Merged
Conversation
…ode serves any queue Phase 6 lands multi-node quorum queues (broker.md §8): - cluster/fabric.rs: one multiplexed TCP connection per peer pair carrying every Raft group's traffic (the shared-transport half of the multi-raft manager) plus the forwarded data plane. Correlation-id RPC (pipelined, no head-of-line blocking, cancellation-safe by construction), bincode Raft payloads, message bodies as the raw frame tail (never serde'd, sliced as refcounted Bytes), batched single-flush writes. - cluster/node.rs: catalog-driven quorum declaration with deterministic rendezvous placement recorded in the catalog; StartGroup fanout with lazy member healing; MetaWrite leader forwarding; the leader-side bridge that lands forwarded publish/subscribe traffic on unmodified queue actors (a per-connection ordered forwarder preserves producer FIFO and keeps the frame reader unblockable). - proxy.rs: the origin-side leader-following proxy — speaks the queue-actor mailbox protocol so the connection driver is untouched; re-resolves on failover, migrates subscriptions re-arming outstanding demand, retries in-flight publishes with epoch-guarded rebinds (one rebind per failover, not one per publish). - quorum actors now exit on demotion (a follower must never dispatch); proxies detect the closed mailbox and follow the new leader. - Clustering is opt-in: BrokerConfig::cluster / ramqp-brokerd --node-id --cluster-listen --seed. Standalone brokers are byte-for-byte unchanged. - supersedes cluster/tcp.rs + cluster/bootstrap.rs (folded into the fabric). Proven kill-the-leader-mid-stream at three levels: single-replica group, 3-node fabric (proxies only), and client-facing e2e with the unmodified ramqp client — zero accepted-message loss in all three. ramqp-broker 0.1.10 -> 0.2.0.
…ssages requeued at close Repeated bench runs against one broker degraded 95µs → 1.7ms → 5ms p50. Root cause chain, found while standing up the quorum-queue bench: 1. The connection driver's close-time settlement drain used FuturesUnordered::next().now_or_never(). The settlement futures await tokio oneshots, whose polls consult the task's COOPERATIVE BUDGET — exhausted right after a busy close — so every poll reported Pending regardless of actual readiness and the drain forwarded NOTHING. 2. The dropped settlements left their messages unacked at the queue; the teardown unsubscribe then requeued them — hundreds of already-accepted messages redelivered to the next consumer as duplicates, every busy close. The drain now runs under tokio::task::unconstrained (bypassing the budget) with a per-item timeout so a peer that really left deliveries unsettled cannot stall teardown (those still requeue — at-least-once, as before). Also forwards resolved settlements in batches during steady state instead of one per wakeup. Regression test: blast + ranged accepts + close, then assert a fresh consumer receives nothing. Diagnostics that found it stay: core gains Session::sender_backlog() / UnsettledMap::id_range(), and bench-compare gains a drain bin. ramqp-core 0.2.1 -> 0.2.2 (additive diagnostics); ramqp-broker 0.2.0 -> 0.2.1.
…lete bench-compare/README gains the quorum matrix (256 B, same harness/machine): p50 288µs leader-local / 427µs via follower (one fabric hop ≈ +140µs), p99.9 < 1ms and p99.9/p50 ≈ 2x on every leg, 140-165k msg/s — vs RabbitMQ 4.3.1 quorum at 2234µs p50 / 39k msg/s. Stated caveats: our Raft log is in-memory vs RabbitMQ's fsync (durability parity re-run lands with Phase 7's on-disk log), and our leg replicated x3 while RabbitMQ ran a single member. broker.md: Phase 6 checked off in full.
…up-commit fsync Phase 7 slice 1 (broker.md): - store.rs (behind the new store-redb feature): one redb database per node; every mutation rides a single group-commit writer thread that drains a burst, applies one write transaction, and fsyncs once — one fsync amortizes every publish in flight across all durable queues (§3.2 batching). Publishes are confirmed only after their batch commits, so the accepted disposition is a real on-disk durability confirm. Reads (recovery scans, dispatch-time body fetches) use MVCC read transactions and never block the writer. - durable.rs: the queue actor over the store — same mailbox contract as the transient/quorum actors, pipelined commits, ready-set dispatch seeded by a recovery scan (restart recovery = the seed; at-least-once). - /durable/<name> addressing; BrokerConfig::data_dir; brokerd --data-dir / RAMQP_DATA_DIR. Without the feature or a data dir the attach is refused (link-level not-found; the session survives). A failed store open (e.g. a stale file lock) is retried on the next attach, not cached. - Client-facing tests: round trip, FULL BROKER RESTART recovery (unsettled messages recovered in FIFO order, settled ones never resurrect), kind-coexistence, and the no-data-dir refusal. ramqp-broker 0.2.1 -> 0.3.0.
…o disk Phase 7 slice 2, the broker.md §8 #1 risk (deep queues must not live in RAM): - cluster/paging.rs: per-queue append-only segment files. Bodies beyond a resident budget (BrokerConfig::resident_bytes_max, default 64 MiB) spill at apply time — the index stays resident, the bytes do not. Segments reclaim when everything in them is settled; snapshot builders pin segments so concurrent settles cannot delete one mid-read. - QueueState is now paging-aware: StoredBody::{Resident,Spilled}, spill on enqueue past the budget, release on settle, dispatch reads spilled bodies back outside the store lock (BodyFetch). - ReplicatedState grows snapshot hooks (prepare/snapshot_bytes/restore) so a paged queue's snapshot keeps spilled bodies EXTERNAL (refs — a million-deep queue's snapshot no longer materializes its payload) and SharedStore parks snapshot blobs on disk when a data_dir exists (a deep snapshot must not double RSS). Cross-node install of an external-body snapshot fails loudly: follower catch-up via snapshot on deep paged queues is the documented follow-up (log-replay catch-up works). - Quorum snapshot cadence 5k -> 50k applies: each build clones+serializes the full index, so at depth the old cadence was 10x the stalls and write amplification. Incremental snapshots are the standing follow-up. - New depth bench (bench-compare/depth) + results in the README: p50/p99 FLAT from empty to 1M deep (117->125µs / ~250µs); at 4 KiB bodies paging cuts RSS ~3x (878 vs 2461 MiB) at identical p50; drain-out from disk 248k msg/s. ramqp-broker 0.3.0 -> 0.4.0.
…ad-lettering
Phase 7 slice 3 (broker.md):
- BrokerConfig::policies: (name-prefix, QueuePolicy) pairs, first match
wins. QueuePolicy: message_ttl, max_length + overflow behavior
(RejectPublish | DropHead), dead_letter target address,
max_delivery_attempts.
- Enforced in all three queue actors (transient, durable, quorum):
- Lazy head-of-queue TTL expiry (RabbitMQ-classic semantics). Quorum
queues stamp the enqueue time INTO the log entry so every replica
agrees; durable queues persist it (store schema now carries
enqueued_ms); requeues keep the original clock.
- DropHead displaces the oldest ready message (dead-lettering it) to
admit the new publish; RejectPublish keeps the existing refuse path.
- A message failing delivery max_delivery_attempts times dead-letters
instead of redelivering forever (poison-message guard).
- One broker-wide dead-letter router: actors emit (target, body); the
router resolves through the registry and republishes pre-settled,
best-effort. It holds the registry WEAKLY — a strong ref kept a
shut-down broker's redb store (and its file lock) alive, breaking
restart recovery; caught by the durable restart test.
- Client-facing tests: TTL->DLX, drop-head ordering, delivery-limit->DLX,
and quorum TTL through the replicated log.
ramqp-broker 0.4.0 -> 0.5.0.
…e restarts Phase 7 complete (broker.md): - cluster/store.rs: SharedStore gains write-through persistence via a new RaftLogSink trait — log appends, votes, conflict truncations, purges, and snapshot pointers are durable BEFORE the Raft storage call returns (the Raft safety requirement). RaftPersistFactory opens per-group sinks + recovery. - store.rs (store-redb): the sink implementation on the same group-commit writer thread — raft appends from every queue group AND the metadata group share one fsync with durable-queue publishes. New tables: raft_groups / raft_log / raft_meta. - Recovery: groups rebuild from persisted vote+log+snapshot at creation; openraft replays committed entries; the quorum actor now WAITS for replay to finish before seeding its ready-set (else recovered messages strand). Spill segments survive restarts (open_preserving) with live counts recomputed from the restored state; unreferenced segments reclaim. - Guard rails: with a data dir, an unopenable store (e.g. a lingering file lock) REFUSES the quorum attach / fails the clustered bind loudly instead of silently starting empty over persisted state — both were real silent-fallback bugs the restart tests caught. - Client-facing tests: standalone quorum restart and single-node-cluster (metadata catalog) restart — unsettled messages recover in FIFO order, settled ones never resurrect, recovered queues accept new work. ramqp-broker 0.5.0 -> 0.6.0. Phase 7 (durability & deep-queue scaling) is complete.
Phase 8 complete (broker.md):
- Coordinator links: a sender attach targeting Coordinator binds a
control link; declare/discharge control messages (amqp-value sections)
are decoded and answered per spec — declare settles with the
declared{txn-id} outcome, discharge of an unknown txn is rejected.
- Transactional enqueues: a transfer carrying transactional-state stages
its message under the txn (settled back with the provisional
transactional outcome) — nothing reaches a queue before discharge.
- Transactional settlements: a consumer disposition carrying
transactional-state stages its ack; the settlement pipeline gained a
SettleAction (apply-now vs stage) split. Coordinator controls drain
ready settlements first so a pipelined discharge cannot take an
incomplete transaction.
- Commit publishes every staged enqueue through its queue's own confirm
(Raft commit for quorum queues, fsync for durable — the coordinator is
cluster-aware by construction), then applies staged settlements;
rollback drops enqueues and requeues settlements. Connection close is
an implicit rollback (local-transactions scope). Staging is bounded
(64 txns/conn, 10k ops/txn).
- Client: Consumer::settle_in_txn (behind the transaction feature);
core: declared_state/txn_state helpers.
- Six client-facing tests incl. atomic visibility, both discharge paths,
quorum-queue commits, and implicit rollback.
ramqp-core 0.2.2 -> 0.2.3; ramqp-broker 0.6.0 -> 0.7.0.
…cs endpoint Phase 9 complete (broker.md): - SCRAM-SHA-1/-256/-512 server flow (full RFC 5802 exchange, server-final signature in the outcome's additional-data — mutual auth against the unmodified client). Authenticator grows scram_verifier() backed by verifier storage (salted+iterated, no plaintext at rest); StaticScram is the built-in store. - Per-address authorization: authorize(identity, vhost, address, operation) is consulted at every link attach BEFORE queue resolution — an unauthorized attach cannot auto-declare; refusals are link-level (unauthorized-access; the session survives). The txn coordinator authorizes as $coordinator. - Vhosts: open.hostname = vhost:<name> namespaces every queue this connection touches as <vhost>/<name> — names, storage, catalog, policies, and permissions all scope; the same address in two vhosts is two queues (tested). Dead-letter targets compose across the scheme. - Management endpoint (BrokerConfig::management_listen / brokerd --management-listen): a dependency-free HTTP server exposing GET /metrics (Prometheus text: connections, RSS, per-queue ready/unacked/consumers) and GET /queues (JSON). Stats are pulled from queue actors at scrape time via a Stats mailbox message — nothing rides the hot path. - Tests: SCRAM round trip (right/wrong password, unknown user, PLAIN refused), authorization gating with session survival, vhost isolation, metrics + inspection. ramqp-broker 0.7.0 -> 0.8.0.
… — Phase 10 core
Phase 10 (broker.md), the in-process legs:
- fe2o3-amqp interop (bench-compare/tests/fe2o3_interop.rs): an
INDEPENDENT AMQP 1.0 implementation against ramqp-broker — handshake,
links, transfers, dispositions incl. release/redelivery, and quorum
queues. All pass unmodified.
- Fault injection: rolling leader kills to the availability boundary
(2/3 alive: recovers and accepts; 1/3: quorum lost, publishes cleanly
REFUSED — never silently accepted, never hung) and follower-loss
transparency (zero refusals, zero loss).
- Two real bugs the fault tests flushed out, both fixed:
1. A dying node's still-open fabric connections could lazily resurrect
an EMPTY group member whose conflict replies below the leader's
matched index panic openraft ("follower log reversion"). Nodes now
refuse member creation once stopping.
2. With zombie resurrection gone, proxies on a stopping node spun their
full 60s rebind window, pinning the node (and its store lock) past
restart. Proxies now abort rebinding when their node is stopping.
- Runtime-model decision (§3.3) recorded in broker.md: STAY on tokio
work-stealing — targets met without escalation (p99.9/p50 ≈ 3-5x vs
the ≤10x bar; tails flat to 1M deep); revisit triggers documented; the
dispatch layer stays shard-partitioned so escalation remains cheap.
- broker.md Phase-10 checkboxes updated with the honest remainder:
Qpid/JMS interop legs, a unified conformance harness, process-level
partition testing, tuned-incumbent benchmark re-runs.
ramqp-broker 0.8.0 -> 0.8.1.
…t-Phases-0-10 - Phase 8 header/checkbox marked complete (the earlier edit missed on a text mismatch). - §14 risks retagged by landed state: deep-queue latency, runtime-model fork, leader-routing fabric, semver, persistence engine, and driver de-dup are retired; the no-GC gate (tuned-incumbent re-run), multi-raft batched ticks at scale, and Jepsen-grade partition testing stay live. - New follow-up items recorded: incremental snapshots, spill-segment shipping for follower snapshot catch-up, fabric authentication, and a richer management/admin surface.
Contributor
Author
|
Automated six-agent review of this branch is in #19 — 4 Critical, 10 High, 18 Medium, 17 Low, deduplicated across cluster/consensus, durability (×2), transactions, security, and policy reviewers. Several findings were reported independently by two agents (flagged ⭐). Net: the consensus ack path and the core protocol paths verified sound; the weight is in recovery/restart paths (silent data-loss windows) and a few DoS / tenant-isolation judgment calls. No fixes applied — triage backlog with stable finding IDs for follow-up PRs. |
…#19) Commit previously published staged enqueues sequentially and broke on the first refusal: enqueues that already landed were never withdrawn while the client was told 'rolled back' — a retry duplicated them. Nothing validated at stage time protected commit time, and a connection dying mid-commit dropped the commit future between publishes. - Two-phase commit: QueueMsg::{Reserve, Unreserve, PublishReserved} across the transient, durable, quorum, and proxy actors (plus matching fabric RequestKinds routed to the leader actor). Every deterministic refusal — full queue, dead/deleted actor — now aborts before a single message lands. - Discharge execution is detached (tokio::spawn): a connection dying mid-commit no longer strands a half-applied transaction. - Non-deterministic mid-apply failures (fsync error, Raft leadership loss) release the remaining reservations and report DischargeOutcome::Partial — the client is told how many enqueues committed instead of a false 'rolled back'. Tests: reserve-protocol unit tests (transient actor), commit-atomicity unit tests (full queue / dead actor / happy path), and a client-facing regression test proving a commit spanning a full queue applies nothing.
…a catalog survives restart (#19) The metadata group ran with the default snapshot policy (LogsSinceLast 5000) and snapshot_dir = None: after enough catalog writes openraft built a Memory-blob snapshot — which save_snapshot skipped (File blobs only) — then DURABLY purged the log behind it. On restart, recovery found the purge marker and log tail but no snapshot: the replicated queue catalog silently reverted or vanished, stranding queue data whose placement could no longer be resolved. - RaftLogSink::save_snapshot now takes SnapshotPersist::{Inline, File}: memory-held blobs (meta catalog, unpaged queues) are stored inside redb, atomic with the snapshot pointer; paged-queue file blobs keep recording their path. - recover_raft returns the inline blob (snap_blob key) or the path (snap_path); exactly one survives per save, atomically. - new_persistent restores from either form. Regression test: 200 catalog writes under an aggressive snapshot policy, wait for the durable purge, restart the store + raft, assert the full catalog (including the purged prefix) recovers.
…ability ordering (#19) Four interlocking holes let a crash or power cut turn a durably-purged log into silent state loss or corruption: - CRIT-4: snapshot blobs were written with std::fs::write (no fsync, no directory fsync) and spill segments with bare write_all, while the snapshot pointer and log purge WERE durable. After power loss the pointer and purge survive but the blob may be missing/short/zero-filled — silent loss or corrupt bodies. Blobs now go through write_blob_durably (write + file fsync + dir fsync); paged snapshot builds fsync every spill segment (Spill::sync_all) before the snapshot pointer persists, since External refs make those bytes the only copy once the log purges. - HIGH-8: install_snapshot deleted the OLD blob before durably recording the new pointer; a crash between the two left recovery pointing at a deleted file. Deletion now waits until the new pointer is durable, and a failed blob write degrades to inline persistence (old file untouched) instead of deleting the old blob with nothing persisted. - HIGH-9: recover_raft silently degraded a missing blob file to 'no snapshot', recovering an empty state below the purge marker. It now refuses to start, loudly, for a dangling pointer or a pointer with no blob at all. - LOW-10: the snapshot-id counter reset to 0 on restart, so a post-restart build at the same applied index truncate-overwrote the live blob in place. The counter now resumes from the recovered snapshot id.
…te, keeping paging (#19) install_snapshot rebuilt the state machine from QueueState::default() (paging: None): - A deep-queue snapshot with PortableBody::External refs made the follower's restore fail ('no spill') → StorageError → the member's Raft core went Fatal, permanently (snapshot-based catch-up structurally broken for deep queues). - An all-inline snapshot installed 'successfully' but silently replaced the paged state with an unpaged one: the replica never spilled again (unbounded RSS) and the old state's spilled segments leaked without release(). The restore now happens in place on the existing state (as new_persistent already did), which preserves the node-local paging config and releases old bodies as they are replaced. Additionally, each spill store now carries a persistent identity (spill.id) stamped into snapshots next to their External refs: restoring a snapshot built against ANOTHER replica's spill — where the same (segment, offset) names different bytes — is rejected loudly instead of silently serving corrupt bodies. Deep-queue snapshot installs remain node-local by design (the standing incremental-snapshot follow-up); what changes is that every misuse now fails loudly instead of corrupting. Tests: install preserves paging (post-install enqueues still spill within the resident budget); foreign-spill snapshots are rejected with a cross-replica error.
… subscriptions (#19) rebind() cleared its channel maps, dropped its event receivers, and bumped the binding epoch — but never sent Unsubscribe (local) or CloseSub (remote) for the old bindings. A rebind landing on the SAME still-alive leader (the common case: a forwarded publish exceeding CALL_TIMEOUT under backpressure returns Retry and triggers a rebind that resolves the same node) left the old leader-side subs in the round-robin ring: dispatches to them flowed into dropped channels, and the leader marked those messages in-flight under a subscriber nobody owned — stranded until the fabric connection or actor died. Up to ~half the stream could strand on an otherwise-healthy cluster. - rebind() now closes every current downstream binding first: local subs unsubscribe from the leader-local actor (requeueing their in-flights), remote subs close their fabric channels (the leader unsubscribes on receipt). - try_bind's per-sub loop closes the subs the failed attempt had already opened before returning, so bind retries no longer accumulate abandoned leader-side subs. Regression test: a same-leader rebind leaves the actor with exactly one subscriber (previously two — the leaked old binding plus the new one).
…se (#19) The TxnDone commit disposition carried only (channel, handle, delivery_id) with no session identity — unlike Deliver/SettleIncoming, which carry binding_gen for exactly this reason. A client could discharge on channel L, end the session while the commit was still fsyncing/Raft-committing, begin a new session that the allocator hands the same channel L, and start publishing: the late TxnDone would find the NEW session at L and emit a settled disposition for a delivery id that now names an unrelated transfer — a publish spuriously confirmed (or rejected) that its queue never saw. TxnDone now records the SessionId the discharge arrived on (session ids are monotonic per connection, never reused), and the outcome is delivered only if the session at that channel is still the same one; otherwise it is dropped with a debug log, exactly like a stale queue command. The positive path is covered by the existing transaction suite (every discharge disposition now passes the id check); the reuse race itself has no deterministic client-level repro without fault injection.
…f stranding (#19) When stage_settle failed — the transaction already discharged (a disposition racing the discharge frame, trivial with a multi-task client) or the 10,001st staged op hitting the cap — the settle was dropped with only a warn log. The client could not observe it (its disposition was fire-and-forget settled:true), the subsequent commit still answered Accepted, and the message sat in the queue's unacked map with no redelivery timer: invisible to every consumer until the connection tore down, which the 'requeued by teardown' comment wrongly assumed would happen. - A refused settle is handed back and the message requeues immediately (at-least-once: a duplicate is recoverable, an invisible message is not). - A cap refusal also marks the transaction rollback-only: its staged work is incomplete, so discharge now rolls back and answers Rejected instead of committing silently partial work. Tests: unit — the cap refusal poisons the txn and hands the settle back; unknown-txn refusals hand it back too. Integration — a settle arriving after its transaction's discharge requeues and redelivers the message.
… after the copy's fate (#19) Dead-lettering a durable message committed the source's Remove to disk while the dead-letter copy still rode an in-memory mpsc to the router: a crash after the Remove batch fsynced and before the DL target durably stored the copy lost a previously-confirmed durable message from both the source and the DLQ. DeadLetter can now carry a confirm resolved once the copy's fate is known (durably stored by the target, refused, or dropped — the router publishes confirmed dead letters with an ack and resolves the confirm off-task so a slow DLQ never stalls broker-wide dead-lettering). The durable actor's three DL sites (drop-head, delivery-limit, TTL expiry) submit their Remove only after that resolution: a crash in the window now redelivers (and re-dead-letters) the message instead of silently losing it — at-least-once, never loss. Sources with no DL target keep the immediate inline Remove. Test: durable → durable TTL dead-lettering end to end over the confirm-ordered path.
…taged-bytes bound (#19) CRIT-2 (decision: accept + document + warn; fabric auth/TLS is a deliberate deferral): the inter-node fabric port speaks unauthenticated, unencrypted frames — any host with TCP reach can drive every queue this node leads, rewrite the replicated catalog, and inject Raft RPCs (a forged high-term Vote is a cluster-wide liveness DoS). The AMQP port warned about this class of exposure; the fabric port did not. Now: - ClusterNode::bootstrap logs a loud warning when the fabric binds a non-loopback address (covers brokerd and embedded users); - ClusterMemberConfig carries a Security section documenting the isolated-trusted-network requirement; the brokerd docs point at it. HIGH-4: MAX_STAGED bounded op COUNT only — 64 txns × 10k ops × 16 MiB messages let one authorized client pin hundreds of GiB of staged bodies (and staging replenishes producer credit, so the producer never stalls). Staged publish bytes are now bounded per connection (MAX_STAGED_BYTES, 64 MiB): over-budget stagings are rejected via disposition as before, the transaction is poisoned rollback-only (its work is incomplete), and discharge returns the bytes to the budget. Unit-tested.
… key collisions (#19) The storage key collapsed (vhost, name) into '<vhost>/<name>' with no validation on either side: a default-vhost attach to /queues/tenantA/secret landed on the same key as queue 'secret' in vhost tenantA, and vhost 'a' + queue 'b/c' ≡ vhost 'a/b' + queue 'c' — cross- tenant read/write/delete below the authz layer, which sees only the raw (vhost, address) pair and cannot defend against the collapse. - Vhosts are validated when the open arrives (before our open goes out): '/' and control characters refuse the connection. - Client address resolution (resolve_in) refuses names containing '/' or control characters, so no client-chosen name can cross the key separator. - The broker-INTERNAL path (resolve: dead-letter router, management) still accepts qualified names — the documented per-vhost DLX composition ('/queues/<vhost>/dead') keeps working; it is now reachable only from operator-configured policies, not from client addresses. Tests: registry unit (cross-separator/control-char refusals + internal qualified resolution reaching the tenant queue) and client-facing (cross- tenant attach refused, '/'-vhost open fails, tenant's data untouched).
…ost grants on static authenticators (#19) HIGH-7: a clustered node without on-disk Raft persistence restarts with an empty log AND empty vote, so it can re-grant a vote in a term it already voted in — a candidate missing committed entries can win, silently losing acknowledged publishes. The config is kept (tests and ephemeral clusters use it) but is no longer silent: ClusterNode::bootstrap warns loudly, and BrokerConfig::data_dir documents the durability consequence. HIGH-10: no shipped authenticator overrode authorize(), whose default allows everything — an authenticated user (or with AllowAll, an anonymous one) could select ANY vhost and ANY queue; tenant isolation required writing a custom Authenticator. StaticPlain and StaticScram now accept per-user vhost grants (with_user_vhosts): a granted user may only attach within its vhosts, ungrated users keep the old any-vhost behavior, and the trait docs state the sharp edge explicitly. Tests: unit (grants gate authorize for both authenticators) and client-facing (a SCRAM user bound to tenant-a authenticates fine but is refused attaches inside tenant-b).
…ngth_bytes) (#19) The defaults admitted ~16 TiB per queue (max_queue_depth 1M × max_message_size 16 MiB) with no byte bound anywhere — transient and in-memory quorum queues live in RAM, so one anonymous producer streaming large messages could OOM the broker while inside every configured bound. - BrokerConfig::max_queue_bytes (default 1 GiB; 0 disables) and QueuePolicy::max_length_bytes (per-queue override), resolved into EffectivePolicy::max_bytes. - Transient actor: tracks held body bytes (ready + unacked); over-bytes publishes are refused, and drop-head evicts as many of the oldest ready messages as the new body needs. - Durable actor: same accounting (sizes recovered via the scan, which now returns body lengths) — bounds the shared disk, not just RAM. - Quorum actor: resident bytes (spilled bodies excluded — paged queues are disk-bounded by design) plus in-flight proposal bytes; over-bytes refuses (no synchronous drop-head: a removal must commit through Raft before resident bytes fall). - Reserved (transaction) publishes bypass the residual check as with the depth bound; their overshoot is capped by the 64 MiB per-connection staging budget. Tests: byte-bound refusal despite depth room; drop-head evicting multiple messages to fit; store scan length round-trip.
MED-17: each management socket spawned a task looping on read with no timeout and no concurrency cap — a slow-loris client parked tasks forever and a connection flood spawned unbounded tasks (the AMQP path has both guards). Requests now run under a 10s whole-request deadline and a 64-connection semaphore; excess sockets are dropped at accept. MED-18: escape_label/escape_json escaped only backslash and double-quote — a queue name containing a newline injected arbitrary Prometheus sample lines into /metrics (alert spoofing) and produced invalid JSON from /queues. Labels now escape per the text exposition format and drop other control characters; JSON escapes all control characters per RFC 8259. (Defense in depth: HIGH-6 already rejects control characters in client-chosen names at resolution.) Unit test: a hostile name with quotes, newline, CR, and a control byte is neutralized in both formats.
The quorum actor read the failure count from APPLIED Raft state, added
one, and proposed a pipelined Settle{requeue:true} while re-readying the
message immediately. Under a fast nack loop the read stayed stale (the
increments were still in flight), so the delivery limit fired late — or
never, when increments persistently failed to commit (only logged).
Transient and durable queues count exactly; quorum diverged.
The actor now keeps a leader-local failure map, seeded from applied state
on first sight of a message and incremented exactly per nack (it is the
only proposer of increments while it leads). The replicated increment
still rides the log so counts survive failovers approximately
(at-least-once, as before); Ack/Drop and dead-lettering clear the entry.
Test: a fast nack loop against a /quorum/ queue dead-letters after
EXACTLY max_delivery_attempts — no third delivery.
…e guard (#19) MED-2: when a queue's initializer failed, it evicted the init cell from the registry map — but a caller already WAITING on that cell could then run its own init attempt, succeed, and set the now-orphaned cell. The next attach created a fresh cell and a SECOND live actor for the same queue: duplicate delivery over one queue_id for durable queues (and concurrent Raft instances over one persisted group), split-brain for transient. The init closure now verifies, under the map lock, that its cell is still the mapped one before doing any side-effectful work: either it observes the eviction (aborts as Orphaned and retries on a fresh cell) or its success blocks the eviction via the existing get().is_none() guard — both orders are safe, and no duplicate actor is ever spawned. MED-3: the documented catch-all '' policy prefix matches the dead-letter queue itself, so messages expiring in it re-published into it with a fresh timestamp — unbounded retention, or a 100%-CPU router↔actor livelock with drop-head at capacity. A queue whose resolved dead-letter target is itself now has dead-lettering disabled (warned once at declaration); longer operator-configured cycles remain documented as the operator's responsibility. Unit-tested.
… TTL clock semantics (#19) MED-4: a clustered quorum queue dead-letters through whichever node leads when the message dies — the policy and target resolve against that node's local registry, so a transient/durable DLQ scatters across nodes after failovers and a consumer on one node silently misses the rest. A /quorum/ target is leader-routed and therefore cluster-wide. The leader actor now warns loudly when a clustered queue's dead-letter target is node-local, and NodeSettings::policies documents that per-node policies must be kept identical (they are node-local configuration; replicated policies arrive with the management API). MED-5: TTL runs on the wall clock — the only clock that survives restarts, which durable/quorum TTLs require (RabbitMQ-classic semantics). The consequences (a forward clock step expiring the affected backlog at once; inter-node skew shifting expiry on clusters) are now documented on QueuePolicy::message_ttl with the operational guidance (slew, don't step).
…sumer recovery (#19) MED-7: a demote→re-elect pair coalescing into one watch notification read as 'still leader', so the actor kept its pre-demotion ready set while the interim leader's enqueues sat in the shared state machine, undeliverable until the next restart. The actor now records the term it started leading under and exits when the term changes even if it looks like the leader — the respawn re-seeds the ready set from applied state. MED-9: dispatch dropped a subscriber whose connection channel died without a clean Unsubscribe, but requeued only the message whose send failed — its other in-flights stayed owned by a ghost on a live actor forever. All three actors (quorum, transient, durable) now requeue everything the dead subscriber held, exactly like Unsubscribe. MED-11: a 30s replay-wait timeout logged an error and SERVED with a partially-seeded ready set — messages applied after the seed were stranded until the next restart. The actor now exits instead; the proxy/registry evicts the dead mailbox and respawns, retrying the wait. Test: a dead consumer channel (no Unsubscribe) has ALL its in-flight messages redelivered to a fresh consumer.
…outside the lock (#19) MED-8: start_local_member checked the stopping flag once at entry, then did seconds of async work before inserting into the groups map. An inbound Raft RPC lazily healing a member concurrently with stop() could insert a fresh EMPTY member after the drain — a resurrected member on a dying node whose conflict replies below its previously-acked matched index panic the LEADER ('follower log reversion', the exact failure the flag exists to prevent). The flag is now re-checked under the same lock stop() drains under, so every interleaving either observes the stop or is drained by it. MED-10: PeerClient::conn() dialed with no timeout while holding the reconnect mutex — a blackholed peer (SYN drop, the classic failover mode) hung for the kernel's ~2-minute TCP timeout with the mutex held, stalling resolve_queue_leader's WhoLeads sweep and proxy binds behind it, sailing past the 60s REBIND_DEADLINE and 10s meta-write deadline (both checked only between attempts). Dials are now bounded at 5s and happen OUTSIDE the mutex (racing dialers keep the first winner).
…h, discharge starvation, SCRAM oracle (#19) MED-13: drain_ready_settlements looped with a 20ms per-call timeout, so every coordinator control message stalled the whole connection ~20ms (~40ms per transaction, all other links frozen). The coordinator path now uses a zero-wait drain (unconstrained poll bypasses the tokio-coop budget so ready oneshots aren't misread as pending) — the client's pipelined transactional dispositions were already processed synchronously in an earlier frame, so they are ready now or never. The 20ms variant stays for cleanup, where futures may be genuinely mid-flight. MED-14: a coordinator link detach removed only its binding, leaking every open transaction — staged settles pinned messages invisibly and staged publish bytes stayed resident until CONNECTION close, and the MAX_TXNS=64 slots leaked (a client re-attaching its coordinator on error exhausted the table). Detach now rolls back all open transactions (TxnManager::take_all), reclaiming slots, staged bytes, and requeueing staged settles. MED-15: txn_results could starve under the biased select. Commit EXECUTION already runs detached (CRIT-1), so only outcome reporting rides the arm; that arm now self-drains every ready outcome per wakeup like settlements. MED-16: SCRAM was a username-existence oracle — an unknown user got an immediate failure (no server-first challenge) while a known user with a wrong password got a challenge then failed. Unknown users now get a deterministic decoy verifier (stable salt per username from a per-process secret) so the exchange is identical until client-final fails (RFC 5802 §7). Tests: take_all drains all txns and frees the budget; the decoy verifier is deterministic per username and differs between usernames.
…vergence fixes (#19) LOW-1: the quorum actor re-readied a message into the dispatch set whenever its remove-commit failed — including expiry/delivery-limit/drop-head removals whose body already went to the DLX, so the message was consumed AND dead-lettered. Committed::Remove now carries readd_on_failure: an ack removal re-readies (still replicated, at-least-once), a dead-letter removal does not (matches the durable actor, which never reinserts after DLX). LOW-2: documented that quorum FIFO leans on FuturesUnordered insertion-order first-polling — a fragility under openraft core-channel backpressure, not a contractual guarantee (the durable actor submits inline and avoids it). LOW-3: open_sub's deserialize-error path returned without removing the sub or closing the leader-side subscription, leaking both; it now close_sub's before bailing. LOW-6: a spill segment that drained to zero live entries while still the current segment was never deleted (release skips the current segment), so it leaked until process exit under steady traffic just above the resident cap. append now reclaims the outgoing current segment as it rolls.
…wn race, snapshot pin leak (#19) LOW-4: the meta bootstrap initialize loop lacked a Fatal arm, so after stop() shut the meta Raft down it fell into the generic 250ms retry and spun forever. Added the arm (the queue-group loop already had it). LOW-5: an OpenSub spawned task could insert a leader-side subscription into the connection's subs map AFTER the reader-loop teardown had drained it, leaking an actor-side subscriber and its delivery-pump task until the actor exited. The subs map now carries a closed flag set at teardown; a late OpenSub observing it undoes its actor-side subscribe instead of inserting. LOW-16: build_snapshot pinned the spill under the lock (prepare_snapshot) and unpinned inside the spawn_blocking closure (snapshot_bytes) — if the blocking task was cancelled before running (runtime shutdown), the unpin never happened and deferred segment deletions were withheld until process exit. Unpin moved to a dedicated finish_snapshot hook the builder ALWAYS calls after the join (the clone shares the spill Arc with the live state, so unpinning through the live state balances the pin on every path).
…h, declare slot leak (#19) LOW-7: a failed rebind in the proxy's Subscribe arm was swallowed (let _ = self.rebind()), so after a >60s leaderless window the proxy stayed alive but permanently unbound — no select arm could re-trigger a bind, and attached consumers starved silently. handle_msg now returns whether to keep running; a rebind that exhausts its deadline closes the proxy so the registry evicts and re-declares it. LOW-8: if the queue actor died between the Subscribe send and its reply, accept_attach returned AFTER accept_peer_attach had already answered the client — a zombie link that ignored flow/drain while the consumer waited forever. It now ends the session with an error so the client learns the link is dead. LOW-14: handle_txn_control's Declare allocated a MAX_TXNS slot via txns.declare() BEFORE checking the session existed; a declare on a since-gone session leaked the slot until connection close. The session is now confirmed live first.
…#19) The coordinator conflated transaction errors onto generic conditions: an unknown-txn discharge answered amqp:not-found and unknown-txn/cap staging both answered amqp:resource-limit-exceeded, and a declare carrying a global-id (a distributed-transaction request) was silently accepted as a local declare. - ramqp-core: added the amqp:transaction: error domain (TransactionError: unknown-id / rollback / timeout) and wired it into ErrorCondition (decode, as_str, From). Bumped to 0.2.4 (additive). - Unknown-txn discharge and unknown-txn staging now answer amqp:transaction:unknown-id; stage_publish distinguishes UnknownTxn from Capped so a cap refusal keeps amqp:resource-limit-exceeded. - A declare with global-id is rejected with amqp:not-implemented (distributed transactions unsupported) instead of becoming a local txn. Tested: stage_publish returns UnknownTxn vs Capped vs Staged.
…ore-abort log, PLAIN docs (#19) LOW-9: per-queue spill/snapshot directories were named by a 64-bit FNV-1a of the queue name. FNV is not collision-resistant, so a client that can declare queues could construct a name colliding with an existing durable queue's tag and share its directories — and Spill::open's remove_dir_all would then destroy the victim's segments (a chosen-input data-destruction vector). Directory tags are now a SHA-256 of the name (via ramqp-core's already-linked scram hash), making a chosen collision infeasible. (rendezvous placement keeps FNV — it is a scoring function, not a security boundary.) LOW-11: Remove/Fail store ops carry no completion channel, so a batch-commit abort silently un-persisted a settle → the message redelivered after restart (at-least-once) with no trace. The writer now logs a warning naming the count of unsignalled removals/fails when their batch fails. LOW-17: documented that StaticPlain is a dev/testing helper — plaintext at rest, and its constant-time compare still leaks password length via the length-mismatch short-circuit; production should use StaticScram or a hash-backed authenticator.
…gression guard (#19) LOW-12: max_queues was a single broker-wide pool, so one tenant could exhaust it and starve every other vhost's new-queue attaches (and the DLX router's auto-declares). Added BrokerConfig::max_queues_per_vhost (default 10k; 0 disables), enforced per non-default vhost at declaration. Tenant A hitting its cap no longer affects tenant B. LOW-13: documented the invariant on the knows_link attach branch — it routes to handle_link_frame, the one attach path that skips authorize(), and stays safe only because a known link is already bound (no new binding or queue resolution). Added a regression test: re-attaching a link whose name was already used for a denied attach stays denied.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Completes the remaining broker.md implementation work: Phases 6–9 in full, plus Phase 10's in-process legs. 11 commits, each a self-contained sub-phase with its own tests and docs.
Phase 6 — leader routing + forwarding fabric (clustering complete)
cluster/fabric.rs— one multiplexed TCP connection per peer pair carrying every Raft group's traffic (shared-transport half of the multi-raft manager) plus the forwarded data plane. Correlation-id RPC (pipelined, no head-of-line blocking, cancellation-safe), bincode Raft payloads, message bodies as raw frame tails (never serde'd), batched single-flush writes.cluster/node.rs— catalog-driven quorum declaration with rendezvous placement recorded in the catalog, StartGroup fanout + lazy member healing, MetaWrite leader forwarding, leader-side pub/sub bridging onto unmodified queue actors.proxy.rs— leader-following local actor speaking the queue-mailbox protocol (connection driver untouched); re-resolves on failover, migrates subscriptions re-arming outstanding demand, epoch-guarded publish retries.Phase 7 — durability & deep-queue scaling
/durable/<name>queues on redb (store-redbfeature,--data-dir): group-commit fsync is the publish confirm; full restart recovery.RaftLogSink): quorum queues and the metadata catalog survive full broker restarts.Phase 8 — transactions
amqp:coordinatortarget (spec part 4): declare/discharge, transactional enqueues + settlements staged per connection, commit through each queue's own durability confirm (cluster-aware by construction), rollback + implicit rollback on disconnect. Client gainsConsumer::settle_in_txn.Phase 9 — auth, limits, management
StaticScrambuilt-in.authorize(identity, vhost, address, operation)at every attach, before queue resolution; link-level refusals.open.hostname = vhost:<name>namespaces queues, storage, catalog, policies, permissions.GET /metrics(Prometheus) +GET /queues(JSON), stats pulled from actors off the hot path.Phase 10 — interop, faults, runtime decision (in-process legs)
fe2o3-amqp(independent AMQP 1.0 implementation) interop tests pass against our broker.Notable bugs found & fixed (each with a regression test)
Remaining (tracked in broker.md §11)
Qpid/proton + JMS interop legs, a unified conformance harness, process-level partition/split-brain injection, tuned-incumbent benchmark re-runs; standing follow-ups: incremental snapshots, spill-segment shipping for follower snapshot catch-up, queue delete/admin API.
Versions: ramqp-broker 0.1.10 → 0.8.1, ramqp-core 0.2.1 → 0.2.3 (additive), ramqp client 0.8.0 (one additive feature-gated method). All tests green under default and
--all-features; clippy/fmt/rustdoc clean.