feat(mq): MQ effect records — kill-9-durable streams/PEL/DLQ/triggers + WAL recycle plane guard (Wave B stage 2a)#291
Merged
Conversation
Adds three new WalRecordType discriminants (MqPush 0x72, MqPop 0x73, MqTrigger 0x74) and rewrites src/mq/wal.rs as a versioned envelope codec for all five MQ mutation kinds (MqCreate, MqPush, MqPop, MqAck, MqTrigger). MqCreate (0x70) and MqAck (0x71) keep their existing discriminants but move to a new payload layout: every payload now starts with a version byte and carries the affected db index explicitly, fixing a pre-existing db-0 hardcode bug. Precedent for changing an on-disk payload layout without bumping the discriminant: WalRecordType::XactCommit's own doc comment notes the 0x51->0x53 format freeze — WAL v3 segments are short-lived and rotated with no cross-version compatibility contract. Every decoder returns None on a malformed payload OR an unsupported (future) version byte, never panics — callers are expected to skip-and-warn rather than abort a replay scan on one bad record. MqPop's payload carries the full claim set (id + delivery_count per claimed message) plus any DLQ routing decisions (source id -> assigned DLQ id), so a later replay can reconstruct the consumer group's PEL and last_delivered_id exactly instead of approximating via a count-based heuristic. ~35 unit tests cover roundtrips, empty/malformed/truncated payloads, and unknown-version rejection for all five record kinds. This is durability-plane scaffolding only; no emission or replay call sites are wired up yet (next commits). author: Tin Dang
Wires the new versioned MQ WAL codec (src/mq/wal.rs) into every owner-shard MQ command handler in src/shard/mq_exec.rs: - handle_create: MqCreate now carries db_index (was hardcoded to 0). - handle_push: builds the MqPush payload inside the with_shard_db closure, before `fields` is moved into `stream.add`, then emits it after the closure returns. - handle_pop: threads a WAL payload out of every early-return path (restructured to return `(Frame, Option<Vec<u8>>)`); records the full claimed-id set with delivery counts, the resulting last_delivered_id, and any DLQ routing decisions (source id -> assigned DLQ id) so replay can reproduce POP's outcome exactly instead of re-deriving it. - handle_ack: emits one MqAck record per acked id (by id, not by count — the replay-side fix landing in the next commit depends on this). - handle_trigger: emits MqTrigger before the registry insert consumes the key/callback bytes. `wal_append_on_slice` moves from private to `pub(crate)` since a later commit calls it from the TXN materialization hop (handler_monoio/txn.rs, handler_sharded/txn.rs, spsc_handler.rs) as well. No replay-side changes yet — these records land on disk but are not yet applied on restart (next commit). author: Tin Dang
…ount
Rewrites replay_mq_wal (src/shard/shared_databases.rs) to apply every MQ
WAL record one at a time, in true on-disk LSN order, via a new
apply_mq_wal_record dispatcher and five apply_mq_* handlers (create, push,
pop, ack, trigger) — replacing the old behavior that collected replayed
state into intermediate maps and rolled the consumer group's PEL back to a
COUNT-based snapshot cursor. That heuristic was wrong by construction: a
POP that padded its claim beyond the client-visible count (MAXDELIVERY > 0
over-claims for DLQ candidates), or a partial ACK of a multi-message claim,
could not be reconstructed from a message count alone.
apply_mq_ack now calls Stream::xack by id (idempotent — a replayed ack that
was already applied, or an id that was never actually pending, is a
harmless no-op). apply_mq_pop rebuilds the PEL entry for every claimed id
with its recorded delivery_count, restores last_delivered_id verbatim, and
replays DLQ routing by looking up each source id's field content from
`stream.entries` (already rebuilt by every MqPush applied so far) before
taking a mutable borrow of `stream.groups`, then re-pushing into the
sibling DLQ stream at its original assigned id. apply_mq_create now creates
the stream in the record's own db_index instead of hardcoding db 0.
Every apply path is idempotent by construction (guarded by
`contains_key`/`xack`'s no-op semantics), so replaying a partially-persisted
segment or restarting twice is safe. Decode failures (malformed payload or
an unsupported future version byte) are skip-and-warned via
warn_skip_mq_record + tracing::warn!, tracked in a new MqReplayStats
counter set logged once per shard — never abort the replay scan.
Unit tests replace the old count-based-rollback test with:
- test_replay_mq_wal_restores_full_lifecycle_by_id
- test_replay_mq_wal_dlq_routing_restored
- test_replay_mq_wal_trigger_registered
- test_replay_mq_wal_survives_prior_db_clear (pins the ordering invariant:
replay_mq_wal must run AFTER the AOF-authority db.clear() wipe, which
was already true in main.rs's call order but had no regression test)
- test_replay_mq_wal_unknown_version_skipped_not_fatal
author: Tin Dang
MQ.PUBLISH materializes its queued intents at TXN commit time on two
separate legs — the self-fold path (the committing connection's own shard
owns the queue) and the foreign-shard path (a ShardMessage::MqCommand hop
to the owning shard) — each duplicated across the monoio and tokio
connection handlers. Neither leg emitted a WAL record for the resulting
stream entry, so a durable queue populated exclusively via MQ.PUBLISH
inside a MULTI/EXEC block had no durability at all, independent of the
mq_exec.rs handle_push fix (previous commits) which only covers the direct
MQ.PUSH command path.
Fixes all four call sites:
- src/server/conn/handler_monoio/txn.rs (self-fold leg)
- src/server/conn/handler_sharded/txn.rs (tokio self-fold leg)
- src/shard/spsc_handler.rs (MqTxnMaterialize handler, shared by both
runtimes' foreign-shard leg)
Each site builds the MqPush payload inside the same with_shard_db closure
that assigns the message id and calls `stream.add`, collects the payloads,
then emits them via mq_exec::wal_append_on_slice after the closure returns
— same pattern as handle_push in mq_exec.rs.
author: Tin Dang
…lity Adds tests/crash_recovery_mq_effects.rs::mq_effect_records_survive_kill9, the RED/GREEN proof for the whole Wave B stage 2a change: under `--appendonly yes`, exercises CREATE -> PUSH x5 -> POP(3) -> ACK(2 of 3) on one queue, a second CREATE/PUSH x2/POP(2) pair on another queue that forces immediate DLQ routing (MAXDELIVERY 1), and a TRIGGER registration on a third queue; kills -9 the server; restarts on the same --dir; and asserts stream content (XLEN/XRANGE field values), the consumer-group delivery cursor (POP resumes at msg 4, not a re-delivery of msg 1..3), the PEL restored by id (XPENDING reports exactly msg 3 pending, not all 3 claimed messages), DLQ routing (MQ DLQLEN), and trigger re-arming (a fresh PUSH notifies mq:trigger:<key> with the original callback payload) all survive. Pre-fix, every one of these assertions fails: the AOF-authority db.clear() wipes the keyspace on restart and nothing replayed it back. Measured WAL footprint (release binary, single shard, --appendonly yes): CREATE + 5xPUSH + POP(3) = 7 effect records total 576 bytes on disk (64-byte v3 segment header + records), average ~82 bytes/record — 68 bytes per PUSH (48-byte payload + 20-byte v3 framing/CRC overhead), ~132 bytes for a 3-claim POP (112-byte payload + framing). Adds fuzz/fuzz_targets/mq_wal_record.rs (registered in fuzz/Cargo.toml and both fuzz-pr/fuzz-nightly matrices in .github/workflows/fuzz.yml), fuzzing all five MQ WAL decoders directly against attacker/corruption-controlled bytes — every decoder is contracted to return None on malformed or unsupported-version input, never panic or read out of bounds. Updates tests/crash_recovery_temporal_mq.rs's module doc comment (which previously explained why a live MQ round trip was NOT possible) to point at the new test now that the gap is closed, and fixes two stale references to a unit test name that this change's replay rewrite superseded. CHANGELOG.md [Unreleased] entry added per the Lint gate's requirement. author: Tin Dang
…/MQ/temporal history Adversarial review of the stage-2a branch (verdict SHIP-WITH-FIXES) confirmed a P0: every WAL recycler — autovacuum Pass C in disk-offload mode, the checkpoint protocol's recycle_segments_before, and the admin VACUUM path — deletes sealed segments against an LSN floor that only covers KV pages and the graph store. The workspace/MQ/temporal planes have no snapshot format in ANY mode, so a disk-offload deployment under --max-wal-size pressure permanently lost MQ/WS/temporal history during normal operation — no crash required. Stage 2a raised the blast radius from queue definitions to full stream content/PEL/DLQ/trigger history. Fix is central, in the recyclers themselves (kernel principle: recycle takes the min across planes): segment_holds_plane_history() walks a sealed segment's record-type bytes and any segment holding a plane record is skipped by BOTH recycle_aggressive and recycle_segments_before, regardless of caller. Fail-closed: unreadable file, non-v3 header, unknown (future) record type, or torn tail all keep the segment; a record_len==0 tail is normal zero-padding. Blocked segments are reported via RecycleStats.segments_blocked_plane; the autovacuum disk-offload arm feeds them into reclamation_wal_recycle_blocked_no_checkpoint_total with a rate-limited warn. Pure-KV segments recycle exactly as before (covered by the pre-existing disk-offload test, kept green). Also from the review round: - P2 regression tests: pre-versioning MqCreate/MqAck payloads (written only by dev builds between PR #286 and this branch) fail closed under the new versioned decoders, including the key_len-low-byte==1 collision shapes (1- and 257-byte keys). - P1 filed as task #46 + CHANGELOG caveat: durable MQ streams deleted via generic DEL/UNLINK/FLUSHALL are resurrected with full content by replay — needs an MqDrop tombstone plane (same class as the fixed vector/KV cold-plane resurrection). - CHANGELOG: measured WAL footprint added (~68 B/small PUSH, ~132 B per 3-claim POP, 576 B for the crash-test lifecycle). Red/green: test_recycle_aggressive_keeps_plane_history_segments and test_recycle_segments_before_keeps_plane_history_segments failed against the unguarded recyclers (compile-red on the stats field, then assertion-red), green after; test_pass_c_disk_offload_keeps_plane_segments_and_counts_blocked covers the autovacuum wiring + counter. author: Tin Dang
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughMQ operations now emit versioned WAL v3 effect records, replay reconstructs durable queue state deterministically, WAL recycling preserves sole-copy plane history, and crash-recovery plus decoder fuzz coverage validates the changes. ChangesMQ WAL durability
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant OwnerShard
participant WALv3
participant Replay
participant MQState
Client->>OwnerShard: Execute MQ command
OwnerShard->>WALv3: Append versioned MQ effect record
WALv3->>Replay: Read records after restart
Replay->>MQState: Rebuild stream and consumer state
MQState-->>Client: Serve restored MQ state
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/persistence/wal_v3/record.rs (1)
531-534: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAdd the new discriminants to the
from_u8roundtrip loop.The enum→u8 asserts at Lines 526-528 cover
MqPush/MqPop/MqTrigger, but the u8→enum roundtrip loop still stops at0x71, so the three newfrom_u8arms (Lines 124-126) are never exercised.💚 Include the new tags in the roundtrip loop
for &v in &[ 0x01, 0x10, 0x20, 0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x40, 0x41, 0x42, 0x50, - 0x52, 0x53, 0x60, 0x61, 0x70, 0x71, + 0x52, 0x53, 0x60, 0x61, 0x70, 0x71, 0x72, 0x73, 0x74, ] {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/persistence/wal_v3/record.rs` around lines 531 - 534, Extend the u8→enum roundtrip loop in the relevant record tests to include the discriminants for MqPush, MqPop, and MqTrigger, matching their new from_u8 arms and the existing enum→u8 assertions. Keep the current loop structure and all existing tags unchanged.
🧹 Nitpick comments (1)
src/persistence/wal_v3/segment.rs (1)
129-175: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winScanning the full segment via
fs::readis wasteful on the recycle hot path.
segment_holds_plane_historyloads the entire segment (payloads included) into memory, but only needs the per-record framing (record_len+ the type byte atoffset+12). For a pure-KV segment it reads to EOF every time, and because plane-heavy WALs deliberately stay above--max-wal-size, Pass C keeps firing every ~30s and re-reads the full WAL set on each tick. Consider aBufReaderthat reads the 12-byte framing prefix and seeks past each payload, early-returning on the first plane record.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/persistence/wal_v3/segment.rs` around lines 129 - 175, Update segment_holds_plane_history to avoid fs::read and full-segment buffering: open the path with a buffered reader, inspect each record’s framing prefix to obtain record_len and the type byte at offset+12, then seek past the remaining payload before continuing. Preserve the existing malformed, unknown-type, invalid-header, and read-error fail-closed behavior, while returning immediately when a plane record is found or a zero-padded tail is reached.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/server/conn/handler_monoio/txn.rs`:
- Around line 170-175: Update the WAL append flow around wal_append_on_slice in
the monoio transaction handler and its identical handler_sharded path so
try_send failures are observable rather than discarded. Propagate or explicitly
handle full/disconnected channel errors with backpressure or a defined
transaction-failure policy, ensuring the commit is not acknowledged until the
MqPush record is successfully enqueued.
In `@src/server/conn/handler_sharded/txn.rs`:
- Around line 177-182: Update the transaction handling around
wal_append_on_slice so WAL enqueue failures are observable rather than ignored
after stream mutation. Propagate or explicitly handle full/disconnected channel
errors, applying the existing durability/backpressure policy before reporting
commit success, and ensure failed MqPush enqueue cannot be treated as a
successful commit.
---
Outside diff comments:
In `@src/persistence/wal_v3/record.rs`:
- Around line 531-534: Extend the u8→enum roundtrip loop in the relevant record
tests to include the discriminants for MqPush, MqPop, and MqTrigger, matching
their new from_u8 arms and the existing enum→u8 assertions. Keep the current
loop structure and all existing tags unchanged.
---
Nitpick comments:
In `@src/persistence/wal_v3/segment.rs`:
- Around line 129-175: Update segment_holds_plane_history to avoid fs::read and
full-segment buffering: open the path with a buffered reader, inspect each
record’s framing prefix to obtain record_len and the type byte at offset+12,
then seek past the remaining payload before continuing. Preserve the existing
malformed, unknown-type, invalid-header, and read-error fail-closed behavior,
while returning immediately when a plane record is found or a zero-padded tail
is reached.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: a5ef634e-38b3-4d68-ad06-c3c38ac6667e
⛔ Files ignored due to path filters (1)
fuzz/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (16)
.github/workflows/fuzz.ymlCHANGELOG.mdfuzz/Cargo.tomlfuzz/fuzz_targets/mq_wal_record.rssrc/mq/wal.rssrc/persistence/wal_v3/record.rssrc/persistence/wal_v3/replay.rssrc/persistence/wal_v3/segment.rssrc/server/conn/handler_monoio/txn.rssrc/server/conn/handler_sharded/txn.rssrc/shard/autovacuum.rssrc/shard/mq_exec.rssrc/shard/shared_databases.rssrc/shard/spsc_handler.rstests/crash_recovery_mq_effects.rstests/crash_recovery_temporal_mq.rs
CodeRabbit (PR #291, both txn.rs call sites): wal_append_on_slice used try_send and discarded the result, so a full/disconnected channel could silently drop an MqPush record AFTER the stream mutation was applied — TXN.COMMIT succeeds, restart recovery loses the message, nobody knows. Blocking is not an option here: every caller runs ON the shard thread, which is also the channel's sole consumer (1ms-tick drain in event_loop.rs) — waiting on a full channel would deadlock the very drain that empties it. Overflow requires >4096 records in a single tick, and the mutation cannot be unwound mid-commit, so the correct policy is observability: on try_send failure emit tracing::error! and increment the new reclamation_wal_append_channel_dropped_total INFO counter (# Reclamation section, field count 29 → 30 with test updated). Operators alert on any increase; zero means the durability guarantee held. Failure-policy rationale documented on the helper. author: Tin Dang
7 tasks
This was referenced Jul 13, 2026
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
2 participants
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.
Summary
Wave B stage 2a (task #34) / storage-kernel M1: MQ gets a real durability sub-plane — stream content, consumer-group PEL, DLQ routing, and trigger registrations now survive kill-9.
Root cause of the loss class:
MQ.*intercepts before the generic AOF-logging dispatch path, so under--appendonly yesthe AOF-authority recovery (db.clear()+ rebuild from the AOF manifest) silently discarded every durable stream on every restart.replay_mq_walcouldn't have saved it either: MqAck rolled the PEL back by COUNT (wrong under DLQ over-claim padding or partial ACKs) and every payload hardcoded db 0.What's in here
MqCreate0x70 /MqAck0x71 bumped in place, newMqPush0x72 (captures the ASSIGNED id → outcome-deterministic replay),MqPop0x73 (claimed ids + resultinglast_delivered_id+ DLQ routing decisions),MqTrigger0x74 (registration as opaque data — replay re-arms, never fires).wal_appendchannel at all 6 owner-shard sites inmq_exec.rs+ 3 MQ.PUBLISH TXN-materialization legs (monoio/sharded/spsc).recycle_segments_before, admin VACUUM) deleted sealed segments against an LSN floor that covers only KV pages + graph — WS/MQ/temporal have NO snapshot in any mode, so disk-offload deployments under--max-wal-sizepressure permanently lost plane history with no crash required. The guard is central in the recyclers: any segment holding a plane record is kept (fail-closed on torn/unknown/unreadable), counted inreclamation_wal_recycle_blocked_no_checkpoint_total, rate-limited warn. Pure-KV segments recycle exactly as before.mq_wal_record(all five decoders, both CI matrices); regression tests proving pre-versioning payload shapes fail closed, including thekey_len-low-byte==1 collision cases.Compatibility
0x70/0x71 payload layout is a deliberate break: MQ WAL writes were dead-wired until PR #286 (2026-07-11) —
wal_append_txwas never assigned — so no released binary ever wrote the legacy layout; only ~1-day-old dev builds did, and those payloads fail closed to skip-and-warn (tested).Known limitation (tracked as task #46 + CHANGELOG caveat)
Durable streams deleted via generic
DEL/UNLINK/FLUSHALLare resurrected with full content by replay after restart — pre-existing class (no MQ tombstone record yet), same shape as the already-fixed vector/KV cold-plane resurrection. Follow-up designs anMqDroptombstone.Verification
tests/crash_recovery_mq_effects.rs: CREATE → PUSH×5 → POP(3) → ACK(2/3) → forced-DLQ pair → TRIGGER,kill -9, restart on same--dir, asserts content + cursor + PEL-by-id + DLQ + trigger re-arm. 3× consecutive green, both runtimes.-D warningsdefault + tokio matrices, fmt.MOON_BIN): crash suitesmq_effects/temporal_mq/wal_recycle_legacyall green, lib 4187/4187.Scope
Durability plane only. Stage 2b (separate PR): replication emission, replica apply arms, FULLRESYNC registry legs, read-only enforcement for WS/MQ.
Summary by CodeRabbit