From 88ef62e8445f56fe16ca63b52e0673964a87897c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Mon, 13 Jul 2026 11:57:09 +0700 Subject: [PATCH] =?UTF-8?q?fix(persistence):=20MQ=20durable-stream=20tombs?= =?UTF-8?q?tone=20=E2=80=94=20DEL/UNLINK/FLUSHALL/FLUSHDB=20no=20longer=20?= =?UTF-8?q?resurrect=20on=20kill-9=20(task=20#46)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: MQ durable streams live as ordinary keys in the shard keyspace, but replay_mq_wal had no way to represent "this queue was deleted after these pushes". A generic DEL/UNLINK/FLUSHDB/FLUSHALL removed the stream from the live db and the DurableQueueRegistry, but replay unconditionally reapplies every MqCreate/MqPush/... record on boot regardless of that delete — a kill-9 after the delete resurrected the full pre-delete content on the next restart. Same bug class as the already-fixed KV/vector cold-plane resurrection (PR #257), now closed for MQ. Reproduced via the crash-matrix RED cell cross_plane_seeded_red_mq_generic_del_resurrection (kernel M3 brief §1.4), now un-gated (no more harness::red_guard) and green 3/3 consecutive runs, plus a full 44/44 default-GREEN crash-matrix pass. Fix: a new MqDrop WAL v3 record (discriminant 0x75) is emitted whenever a durable MQ stream is removed via generic DEL/UNLINK/FLUSHDB/FLUSHALL. Layout matches the other versioned MQ records: [version:u8][db_index:u32] [key_len:u32][key:N], fails closed on any malformed/future-version payload. New hooks mq_exec::auto_drop_mq_streams / auto_drop_mq_streams_on_flush are wired into every connection-layer write path that already runs the equivalent vector/text index-parity hooks (handler_monoio/mod.rs, handler_sharded/mod.rs, the sharded MULTI/EXEC helper in server/conn/shared.rs) plus the replica-side apply_index_parity_hooks in replication/apply.rs — a replica must tombstone its OWN WAL too, or it resurrects the stream on its own restart even though its live copy stayed correctly deleted via normal command replication. Boot-time replay applies apply_mq_drop (shared_databases.rs) strictly in WAL order alongside every other MQ record, so a Drop only kills records that PRECEDE it for that key — a later MqCreate/MqPush for the same key survives intact (create -> drop -> create round-trips a kill-9 with the second incarnation whole; proven by both a unit test and the new crash-matrix cell cross_plane_mq_create_drop_create_survives). segment_plane_scan's plane-history block set gained MqDrop alongside the other MQ discriminants so autovacuum/recycle never deletes a sealed segment still holding an unfloored tombstone. The MQ WAL fuzz target now also fuzzes decode_mq_drop. Replication decision: MQ effect records replicate live only at num_shards == 1 (the pre-existing gate shared with MqCreate/etc). MqDrop follows the same posture — MQ._REPL.DROP is emitted and applied by replication::apply::apply_mq exactly like the other MQ replay commands. Separately, a replicated generic DEL/UNLINK/FLUSHDB/FLUSHALL already removes the stream from the replica's live keyspace via normal command replication regardless of that gate; what it did not do before this fix is tombstone the replica's own wal-v3 MQ plane, which apply_index_parity_hooks now closes. Scope decision (documented in code): DurableQueueRegistry entries are NOT db-indexed (pre-existing limitation, same as MqCreate's registry) — a key match tombstones regardless of which db the deleting command ran in, and FLUSHDB drops every registered durable queue exactly like FLUSHALL since the registry cannot scope to one db. Two different dbs sharing an MQ queue NAME is not a supported configuration. Test-gotcha fixed along the way: the crash-matrix DEL/FLUSHALL scenarios' original sync-marker strategy waited on an AOF-family write to prove the delete was durable — correct pre-fix (DEL only touched the AOF), but MqDrop lands on the wal-v3 MQ plane via a separate fire-and-forget channel drained on its own 1ms tick, so an AOF-only marker no longer proves the tombstone itself reached disk. Both tests now also sync a throwaway durable queue's MQ.PUSH (wal-v3-family) after the delete/flush before crashing. Files touched: - src/mq/wal.rs — encode_mq_drop/decode_mq_drop + MQ_REPL_DROP + is_mq_replay_command update + module docs + unit tests - src/persistence/wal_v3/record.rs — WalRecordType::MqDrop = 0x75 - src/persistence/wal_v3/replay.rs — MqDrop routed through on_command - src/persistence/wal_v3/segment.rs — MqDrop added to plane-history block set - src/shard/shared_databases.rs — apply_mq_drop + replay dispatch wiring + MqReplayStats.drop + 4 new unit tests - src/shard/mq_exec.rs — auto_drop_mq_streams / auto_drop_mq_streams_on_flush / emit_mq_drops - src/server/conn/handler_monoio/mod.rs, src/server/conn/handler_sharded/mod.rs, src/server/conn/shared.rs — hook call sites (DEL/UNLINK + FLUSHDB/FLUSHALL) - src/replication/apply.rs — MQ_REPL_DROP apply arm + apply_index_parity_hooks replica-side tombstone - fuzz/fuzz_targets/mq_wal_record.rs — fuzz decode_mq_drop - tests/crash_matrix_cross_plane/tests_seeded_red.rs — un-gated the RED cell, fixed its wal-v3-family sync race, added a FLUSHALL sibling and a create->drop->create ordering test - CHANGELOG.md — new [Unreleased] entry + removed stale #46 caveats author: Tin Dang --- CHANGELOG.md | 87 ++++++- fuzz/fuzz_targets/mq_wal_record.rs | 10 +- src/mq/wal.rs | 100 ++++++++ src/persistence/wal_v3/record.rs | 12 +- src/persistence/wal_v3/replay.rs | 3 +- src/persistence/wal_v3/segment.rs | 3 +- src/replication/apply.rs | 19 +- src/server/conn/handler_monoio/mod.rs | 7 + src/server/conn/handler_sharded/mod.rs | 15 ++ src/server/conn/shared.rs | 7 + src/shard/mq_exec.rs | 106 +++++++++ src/shard/shared_databases.rs | 223 +++++++++++++++++- .../tests_seeded_red.rs | 163 +++++++++++-- 13 files changed, 711 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cbd2f284..aebc824e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -99,6 +99,75 @@ GREEN. Crash-matrix cells `cross_plane_prod_s1_txn_isolated_committed` / `cross_plane_prod_s4_txn_isolated_committed` (`tests/crash_matrix_cross_plane/`) flip from `red_guard`-gated RED to default-GREEN tripwires (42 cells: 33→35 GREEN by default, 9→7 RED). +### Fixed — MQ durable-stream tombstone: DEL/UNLINK/FLUSHALL/FLUSHDB no longer resurrect on kill-9 (kernel M3 stage 3 / task #46) + +Root cause: MQ durable streams live as ordinary keys in the shard keyspace, +but `replay_mq_wal` had no way to represent "this queue was deleted after +these pushes" — a generic `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` removed the +stream from the live db and the `DurableQueueRegistry`, but a kill-9 +afterward re-materialized the full pre-delete content on the next restart +(replay reapplies every `MqCreate`/`MqPush`/... record regardless). Same +bug class as the already-fixed KV/vector cold-plane resurrection (PR #257), +now closed for MQ. Proven by the former RED crash-matrix cell +`cross_plane_seeded_red_mq_generic_del_resurrection` +(`tests/crash_matrix_cross_plane/tests_seeded_red.rs`), now un-gated (no +more `harness::red_guard`) and green 3/3 consecutive runs. + +Fix: a new `MqDrop` WAL v3 record (discriminant `0x75`, +`src/persistence/wal_v3/record.rs`; versioned encode/decode in +`src/mq/wal.rs` — `[version:u8][db_index:u32][key_len:u32][key:N]`, fails +closed on any malformed/future-version payload like every other MQ record) +is emitted whenever a durable MQ stream is removed via generic +`DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` — new hooks +`mq_exec::auto_drop_mq_streams` / `auto_drop_mq_streams_on_flush`, wired +into every connection-layer write path that already runs the equivalent +vector/text index-parity hooks (`handler_monoio/mod.rs`, +`handler_sharded/mod.rs`, the sharded MULTI/EXEC helper in +`server/conn/shared.rs`, and the replica-side `apply_index_parity_hooks` in +`replication/apply.rs` — a replica must tombstone its OWN WAL too, or it +resurrects the stream on its own restart even though its live copy stayed +correctly deleted). Boot-time replay applies `apply_mq_drop` +(`src/shard/shared_databases.rs`) strictly in WAL order alongside every +other MQ record, so a Drop only kills records that PRECEDE it for that +key — a later `MqCreate`/`MqPush` for the same key survives intact +(create -> drop -> create round-trips a kill-9 with the second incarnation +whole; new regression tests +`test_replay_mq_wal_drop_only_kills_prior_records` and the crash-matrix +`cross_plane_mq_create_drop_create_survives`). `segment_plane_scan`'s +plane-history block set (`src/persistence/wal_v3/segment.rs`) gained +`MqDrop` alongside the other MQ discriminants so autovacuum/recycle never +deletes a sealed segment still holding an unfloored tombstone. The MQ WAL +fuzz target (`fuzz/fuzz_targets/mq_wal_record.rs`) now also fuzzes +`decode_mq_drop`. + +Replication decision: MQ effect records replicate live only at +`num_shards == 1` (the pre-existing gate, matching `MqCreate`/etc — see +`mq_exec::replicate_mq_record`'s docs). `MqDrop` follows the same posture: +`MQ._REPL.DROP` is emitted and applied by `replication::apply::apply_mq` +exactly like the other MQ replay commands. Separately — and regardless of +that gate — a REPLICATED generic `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` +already removes the stream from the replica's live keyspace via normal +command replication; what it did NOT do before this fix is tombstone the +replica's OWN wal-v3 MQ plane, so the replica's live state was correct but +its restart-durability was not. `apply_index_parity_hooks` now closes that +gap identically on the replica side. + +Scope decision (documented in code, not re-litigated per-callsite): +`DurableQueueRegistry` entries are NOT db-indexed (pre-existing limitation, +same as `MqCreate`'s registry) — a key match tombstones regardless of +which db the deleting command ran in, and `FLUSHDB` drops every registered +durable queue exactly like `FLUSHALL` (the registry has no way to scope to +one db). Two different dbs sharing an MQ queue NAME is not a supported +configuration. + +Test-gotcha fixed along the way: the crash-matrix DEL/FLUSHALL scenarios' +original sync-marker strategy waited on an AOF-family write to prove the +delete was durable — correct pre-fix (DEL only touched the AOF), but the +new `MqDrop` record lands on the wal-v3 MQ plane via a separate +fire-and-forget channel drained on its own 1ms tick, so an AOF-only marker +no longer proves anything about the tombstone's own durability. Both tests +now also sync a throwaway durable queue's `MQ.PUSH` (wal-v3-family) after +the delete/flush before crashing. ### Added — unified per-shard floor register + min-across-planes WAL recycle (kernel M3 stage 2 / K2) @@ -283,9 +352,10 @@ fixed — that is out of scope for this stage) for the kernel M3 backlog: Also confirmed GREEN (not RED, contrary to the brief's grouping at the design-doc level): `WS DROP` durability under kill-9 — `WorkspaceDrop` WAL -records replay correctly in order, unlike the sibling MQ generic-`DEL` -resurrection gap (still RED, same bug class as the already-fixed KV/vector -cold-plane resurrection, PR #257, not yet applied to MQ). +records replay correctly in order. The sibling MQ generic-`DEL` +resurrection gap (same bug class as the already-fixed KV/vector cold-plane +resurrection, PR #257) was RED at the time of this entry; **fixed in +kernel M3 stage 3 (task #46)** — see the `MqDrop` entry below. `tests/common/mod.rs` gains `find_moon_binary()`/`sigkill()`/ `wait_for_port_down()` shared helpers this suite depends on (the latter now @@ -577,11 +647,12 @@ crash required. Kept segments are counted in workloads until plane checkpointing lands (storage-kernel M3). The guard fails closed: unreadable/torn/unknown-type segments are kept. -Known limitation (pre-existing, now tracked): durable MQ streams deleted -via generic `DEL`/`UNLINK`/`FLUSHALL`/`FLUSHDB` are resurrected — now with -full content — by MQ WAL replay after a restart; there is no MQ tombstone -record yet (same bug class as the fixed vector/KV cold-plane resurrection; -follow-up task filed). +Known limitation (pre-existing, tracked as task #46): durable MQ streams +deleted via generic `DEL`/`UNLINK`/`FLUSHALL`/`FLUSHDB` are resurrected — +now with full content — by MQ WAL replay after a restart; there is no MQ +tombstone record yet (same bug class as the fixed vector/KV cold-plane +resurrection). **Fixed in kernel M3 stage 3** — see the "MQ durable-stream +tombstone" entry below. Out of scope (stage 2b+): replication emission/apply for the MQ plane. diff --git a/fuzz/fuzz_targets/mq_wal_record.rs b/fuzz/fuzz_targets/mq_wal_record.rs index 8673410a6..723362e91 100644 --- a/fuzz/fuzz_targets/mq_wal_record.rs +++ b/fuzz/fuzz_targets/mq_wal_record.rs @@ -2,13 +2,14 @@ use libfuzzer_sys::fuzz_target; use moon::mq::wal::{ - decode_mq_ack, decode_mq_create, decode_mq_pop, decode_mq_push, decode_mq_trigger, - peek_version, + decode_mq_ack, decode_mq_create, decode_mq_drop, decode_mq_pop, decode_mq_push, + decode_mq_trigger, peek_version, }; -/// Fuzz the MQ WAL v3 op-blob decoders (Wave B stage 2a / task #34). +/// Fuzz the MQ WAL v3 op-blob decoders (Wave B stage 2a / task #34; MqDrop +/// added kernel M3 stage 3 / task #46). /// -/// `data` is fed to all five decoders directly -- exactly what +/// `data` is fed to all six decoders directly -- exactly what /// `src/shard/shared_databases.rs::apply_mq_wal_record` does with a raw WAL /// record payload straight off disk (attacker/corruption-controlled: a /// truncated write, a torn page, or a future-version payload written by a @@ -23,4 +24,5 @@ fuzz_target!(|data: &[u8]| { let _ = decode_mq_push(data); let _ = decode_mq_pop(data); let _ = decode_mq_trigger(data); + let _ = decode_mq_drop(data); }); diff --git a/src/mq/wal.rs b/src/mq/wal.rs index 3c2f72cff..22d018701 100644 --- a/src/mq/wal.rs +++ b/src/mq/wal.rs @@ -9,6 +9,12 @@ //! trigger registrations all survive a kill-9 even when `--appendonly yes` //! would otherwise discard the whole keyspace. //! +//! Kernel M3 stage 3 (task #46) adds `MqDrop` (0x75): a tombstone emitted +//! when a durable stream is removed via generic keyspace `DEL`/`UNLINK`/ +//! `FLUSHDB`/`FLUSHALL` (as opposed to any `MQ.*` command), closing the gap +//! where such a delete had no WAL representation and replay resurrected the +//! full pre-delete content after a kill-9. +//! //! All payloads share a leading `version: u8` byte. Current version is `1` //! for every record kind below. Decoders return `None` for ANY structurally //! invalid payload (including an unrecognized/future version byte) -- NEVER @@ -400,6 +406,47 @@ pub fn decode_mq_trigger(payload: &[u8]) -> Option<(Vec, Vec, Vec, u Some((trig_key, queue_key, callback_cmd, debounce_ms)) } +// ── MqDrop (0x75) ───────────────────────────────────────────────────────── + +/// Encode an MqDrop WAL payload — a tombstone emitted when a durable MQ +/// stream is removed via generic keyspace `DEL`/`UNLINK`/`FLUSHDB`/ +/// `FLUSHALL` (kernel M3 stage 3 / task #46). Replay applies these strictly +/// in WAL order, so a `Drop` only kills the `MqCreate`/`MqPush`/... records +/// that precede it for this `(db_index, key)` pair — a later `MqCreate` for +/// the same key re-materializes normally (create -> drop -> create survives +/// intact). +/// +/// Layout: `[version:u8=1][db_index:u32 LE][key_len:u32 LE][key:N]` +pub fn encode_mq_drop(db_index: u32, queue_key: &[u8]) -> Vec { + let mut payload = Vec::with_capacity(1 + 4 + 4 + queue_key.len()); + payload.push(MQ_WAL_VERSION); + payload.extend_from_slice(&db_index.to_le_bytes()); + payload.extend_from_slice(&(queue_key.len() as u32).to_le_bytes()); + payload.extend_from_slice(queue_key); + payload +} + +/// Decode an MqDrop WAL payload. +/// +/// Returns `(db_index, queue_key)` or `None` if malformed or an unsupported +/// version. +pub fn decode_mq_drop(payload: &[u8]) -> Option<(u32, Vec)> { + if payload.is_empty() || payload[0] != MQ_WAL_VERSION { + return None; + } + let p = &payload[1..]; + if p.len() < 8 { + return None; + } + let db_index = u32::from_le_bytes(p[0..4].try_into().ok()?); + let key_len = u32::from_le_bytes(p[4..8].try_into().ok()?) as usize; + if p.len() < 8 + key_len { + return None; + } + let key = p[8..8 + key_len].to_vec(); + Some((db_index, key)) +} + // ── Replication wire framing (Wave B stage 2b) ──────────────────────────── // // MQ effect records are versioned BINARY payloads, not RESP commands (unlike @@ -424,6 +471,8 @@ pub const MQ_REPL_POP: &[u8] = b"MQ._REPL.POP"; pub const MQ_REPL_ACK: &[u8] = b"MQ._REPL.ACK"; /// Synthetic replication pseudo-command for [`WalRecordType::MqTrigger`]. pub const MQ_REPL_TRIGGER: &[u8] = b"MQ._REPL.TRIGGER"; +/// Synthetic replication pseudo-command for [`WalRecordType::MqDrop`]. +pub const MQ_REPL_DROP: &[u8] = b"MQ._REPL.DROP"; /// True for any `MQ._REPL.*` synthetic replication pseudo-command emitted by /// [`crate::shard::mq_exec`]'s live fan-out. Used by @@ -438,6 +487,7 @@ pub fn is_mq_replay_command(cmd: &[u8]) -> bool { || cmd.eq_ignore_ascii_case(MQ_REPL_POP) || cmd.eq_ignore_ascii_case(MQ_REPL_ACK) || cmd.eq_ignore_ascii_case(MQ_REPL_TRIGGER) + || cmd.eq_ignore_ascii_case(MQ_REPL_DROP) } #[cfg(test)] @@ -451,6 +501,8 @@ mod tests { assert!(is_mq_replay_command(b"MQ._REPL.POP")); assert!(is_mq_replay_command(b"MQ._REPL.ACK")); assert!(is_mq_replay_command(b"MQ._REPL.TRIGGER")); + assert!(is_mq_replay_command(b"MQ._REPL.DROP")); + assert!(is_mq_replay_command(b"mq._repl.drop")); assert!(!is_mq_replay_command(b"MQ")); assert!(!is_mq_replay_command(b"MQ.PUSH")); assert!(!is_mq_replay_command(b"GRAPH.ADDNODE")); @@ -688,6 +740,54 @@ mod tests { assert!(decode_mq_pop(&payload).is_none()); } + // --- MqDrop roundtrip --- + + #[test] + fn test_mq_drop_roundtrip() { + let payload = encode_mq_drop(3, b"orders"); + let (db, key) = decode_mq_drop(&payload).unwrap(); + assert_eq!(db, 3); + assert_eq!(key, b"orders"); + } + + #[test] + fn test_mq_drop_roundtrip_empty_key() { + let payload = encode_mq_drop(0, b""); + let (db, key) = decode_mq_drop(&payload).unwrap(); + assert_eq!(db, 0); + assert!(key.is_empty()); + } + + #[test] + fn test_mq_drop_malformed_empty() { + assert!(decode_mq_drop(b"").is_none()); + } + + #[test] + fn test_mq_drop_malformed_truncated_key() { + let mut bad = vec![MQ_WAL_VERSION]; + bad.extend_from_slice(&0u32.to_le_bytes()); + bad.extend_from_slice(&100u32.to_le_bytes()); // key_len = 100 + bad.extend_from_slice(&[0u8; 10]); // only 10 bytes of key + assert!(decode_mq_drop(&bad).is_none()); + } + + #[test] + fn test_mq_drop_unknown_version_rejected() { + let mut payload = encode_mq_drop(0, b"q"); + payload[0] = 42; + assert!(decode_mq_drop(&payload).is_none()); + } + + #[test] + fn test_mq_drop_extra_bytes_ignored() { + let mut payload = encode_mq_drop(1, b"q"); + payload.extend_from_slice(b"trailing"); + let (db, key) = decode_mq_drop(&payload).unwrap(); + assert_eq!(db, 1); + assert_eq!(key, b"q"); + } + // --- MqTrigger roundtrip --- #[test] diff --git a/src/persistence/wal_v3/record.rs b/src/persistence/wal_v3/record.rs index bc02bf4d1..985a2331e 100644 --- a/src/persistence/wal_v3/record.rs +++ b/src/persistence/wal_v3/record.rs @@ -94,6 +94,14 @@ pub enum WalRecordType { /// durable/replayed as opaque data; the callback is never fired during /// replay (only live `MQ.PUSH` debounce arming fires it). MqTrigger = 0x74, + /// MQ durable-stream tombstone record (kernel M3 stage 3 / task #46). + /// Emitted when a durable MQ stream is removed via generic keyspace + /// `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` — without this, `replay_mq_wal` + /// has no way to represent "this queue was deleted after these pushes" + /// and re-materializes full pre-delete content on every restart. Replay + /// applies it strictly in WAL order, so it only kills PRIOR records for + /// this `(db_index, key)`; a later `MqCreate` re-materializes normally. + MqDrop = 0x75, } impl WalRecordType { @@ -124,6 +132,7 @@ impl WalRecordType { 0x72 => Some(Self::MqPush), 0x73 => Some(Self::MqPop), 0x74 => Some(Self::MqTrigger), + 0x75 => Some(Self::MqDrop), _ => None, } } @@ -526,11 +535,12 @@ mod tests { assert_eq!(WalRecordType::MqPush as u8, 0x72); assert_eq!(WalRecordType::MqPop as u8, 0x73); assert_eq!(WalRecordType::MqTrigger as u8, 0x74); + assert_eq!(WalRecordType::MqDrop as u8, 0x75); // from_u8 roundtrips 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, 0x75, ] { assert!(WalRecordType::from_u8(v).is_some()); } diff --git a/src/persistence/wal_v3/replay.rs b/src/persistence/wal_v3/replay.rs index 2aa7e8e3c..c0ba76ad9 100644 --- a/src/persistence/wal_v3/replay.rs +++ b/src/persistence/wal_v3/replay.rs @@ -416,7 +416,8 @@ pub fn replay_wal_v3_file_until( | WalRecordType::MqAck | WalRecordType::MqPush | WalRecordType::MqPop - | WalRecordType::MqTrigger => { + | WalRecordType::MqTrigger + | WalRecordType::MqDrop => { on_command(&record); result.commands_replayed += 1; } diff --git a/src/persistence/wal_v3/segment.rs b/src/persistence/wal_v3/segment.rs index 266548750..e609b89ef 100644 --- a/src/persistence/wal_v3/segment.rs +++ b/src/persistence/wal_v3/segment.rs @@ -203,7 +203,8 @@ fn segment_plane_scan(path: &Path) -> SegmentPlaneScan { | WalRecordType::MqAck | WalRecordType::MqPush | WalRecordType::MqPop - | WalRecordType::MqTrigger, + | WalRecordType::MqTrigger + | WalRecordType::MqDrop, ) => return BLOCKED, Some(WalRecordType::GraphTemporal) => has_graph_temporal = true, Some(_) => {} diff --git a/src/replication/apply.rs b/src/replication/apply.rs index 409df96be..d071d9c8a 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -444,11 +444,12 @@ fn apply_graph(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Fram /// live via `MQ.PUSH`'s debounce arming on a master). fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame]) { use crate::mq::wal::{ - MQ_REPL_ACK, MQ_REPL_CREATE, MQ_REPL_POP, MQ_REPL_PUSH, MQ_REPL_TRIGGER, decode_mq_ack, - decode_mq_create, decode_mq_pop, decode_mq_push, decode_mq_trigger, + MQ_REPL_ACK, MQ_REPL_CREATE, MQ_REPL_DROP, MQ_REPL_POP, MQ_REPL_PUSH, MQ_REPL_TRIGGER, + decode_mq_ack, decode_mq_create, decode_mq_drop, decode_mq_pop, decode_mq_push, + decode_mq_trigger, }; use crate::shard::shared_databases::{ - apply_mq_ack, apply_mq_create, apply_mq_pop, apply_mq_push, apply_mq_trigger, + apply_mq_ack, apply_mq_create, apply_mq_drop, apply_mq_pop, apply_mq_push, apply_mq_trigger, }; use crate::storage::stream::StreamId; @@ -542,6 +543,10 @@ fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame]) decode_mq_trigger(payload).map(|(trig_key, queue_key, callback_cmd, debounce_ms)| { apply_mq_trigger(s, trig_key, queue_key, callback_cmd, debounce_ms); }) + } else if cmd.eq_ignore_ascii_case(MQ_REPL_DROP) { + decode_mq_drop(payload).map(|(db_index, key)| { + apply_mq_drop(s, clamp_mq_db(db_count, db_index), &key); + }) } else { // Unreachable: `is_mq_replay_command` gated this call. Defensive. tracing::debug!( @@ -624,6 +629,11 @@ fn apply_index_parity_hooks( } } else if cmd.eq_ignore_ascii_case(b"DEL") || cmd.eq_ignore_ascii_case(b"UNLINK") { hooks::auto_delete_vectors(&mut s.vector_store, args, db_index); + // task #46: a replica must also tombstone its OWN WAL for any + // durable MQ stream a replicated generic DEL/UNLINK removed — + // otherwise the replica resurrects it on its own restart even + // though the master's copy stayed correctly deleted. + crate::shard::mq_exec::auto_drop_mq_streams(s, args, db_index as usize); } else if cmd.eq_ignore_ascii_case(b"HDEL") { hooks::auto_hdel_vectors(&mut s.vector_store, args, db_index); } else if cmd.eq_ignore_ascii_case(b"FLUSHDB") || cmd.eq_ignore_ascii_case(b"FLUSHALL") { @@ -633,6 +643,9 @@ fn apply_index_parity_hooks( cmd.eq_ignore_ascii_case(b"FLUSHDB"), db_index, ); + // task #46: same replica-side tombstone requirement as DEL/UNLINK + // above. + crate::shard::mq_exec::auto_drop_mq_streams_on_flush(s, db_index as usize); } } diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 514d5ae28..236abed6b 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1647,6 +1647,10 @@ pub(crate) async fn handle_connection_sharded_monoio< cmd_args, sel_db as u8, ); + // task #46: tombstone any durable MQ stream(s) + // this generic DEL/UNLINK removed, so + // `replay_mq_wal` doesn't resurrect them. + crate::shard::mq_exec::auto_drop_mq_streams(s, cmd_args, sel_db); } // R4: HDEL of an indexed vector field tombstones it. @@ -1671,6 +1675,9 @@ pub(crate) async fn handle_connection_sharded_monoio< cmd.eq_ignore_ascii_case(b"FLUSHDB"), sel_db as u8, ); + // task #46: tombstone every durable MQ stream + // this FLUSHDB/FLUSHALL cleared. + crate::shard::mq_exec::auto_drop_mq_streams_on_flush(s, sel_db); } // Blocking wakeup: re-borrow db by index (NLL) diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index b7628ad94..03e63d36c 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1646,6 +1646,15 @@ pub(crate) async fn handle_connection_sharded_inner< ); } } + // task #46: tombstone any durable MQ + // stream(s) this generic DEL/UNLINK + // removed, so `replay_mq_wal` doesn't + // resurrect them. + crate::shard::mq_exec::auto_drop_mq_streams( + s, + cmd_args, + conn.selected_db, + ); }); } // R4: HDEL of an indexed VECTOR field tombstones the vector @@ -1677,6 +1686,12 @@ pub(crate) async fn handle_connection_sharded_inner< cmd.eq_ignore_ascii_case(b"FLUSHDB"), conn.selected_db as u8, ); + // task #46: tombstone every durable MQ + // stream this FLUSHDB/FLUSHALL cleared. + crate::shard::mq_exec::auto_drop_mq_streams_on_flush( + s, + conn.selected_db, + ); }); // D-2: keyless flush routed local-only cleared just this // shard — broadcast to every other shard so the whole diff --git a/src/server/conn/shared.rs b/src/server/conn/shared.rs index d936ab32a..85d7c0484 100644 --- a/src/server/conn/shared.rs +++ b/src/server/conn/shared.rs @@ -380,6 +380,10 @@ pub(crate) fn execute_transaction_sharded( cmd_args, selected as u8, ); + // task #46: tombstone any durable MQ stream(s) this generic + // DEL/UNLINK removed, so `replay_mq_wal` doesn't resurrect + // them. + crate::shard::mq_exec::auto_drop_mq_streams(s, cmd_args, selected); }); } @@ -406,6 +410,9 @@ pub(crate) fn execute_transaction_sharded( cmd.eq_ignore_ascii_case(b"FLUSHDB"), selected as u8, ); + // task #46: tombstone every durable MQ stream this + // FLUSHDB/FLUSHALL cleared. + crate::shard::mq_exec::auto_drop_mq_streams_on_flush(s, selected); }); } diff --git a/src/shard/mq_exec.rs b/src/shard/mq_exec.rs index be6639197..0b9faedca 100644 --- a/src/shard/mq_exec.rs +++ b/src/shard/mq_exec.rs @@ -297,6 +297,112 @@ fn replicate_mq_record( ) { } +// ── Generic-keyspace tombstone hooks (kernel M3 stage 3 / task #46) ─────────── +// +// MQ durable streams live as ordinary keys in the shard's keyspace, so a +// GENERIC `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` (not an `MQ.*` command) can +// remove one without going through `handle_create`/etc. Before this fix +// `replay_mq_wal` had no way to represent "this queue was deleted after +// these pushes": a kill-9 after such a delete resurrected the full +// pre-delete content on restart. These hooks mirror the vector/text +// index-parity hooks (`auto_delete_vectors` / `auto_flush_indexes` in +// `spsc_handler.rs`) and the WS.DROP `WorkspaceDrop` precedent, and are +// called from every connection-layer write path AFTER the generic command +// already succeeded (same call shape as those hooks) — using DIRECT field +// access on an already-owned `&mut ShardSlice` (never re-enters +// `with_shard`, matching the "Direct field access" convention documented at +// the KV undo-log call site in `handler_monoio/mod.rs`). +// +// Scope decision: `DurableQueueRegistry` keys are NOT db-indexed (the same +// pre-existing limitation `MqCreate`'s registry already has — see its doc +// comment); a key match tombstones regardless of which db the command ran +// in, and `FLUSHDB` drops every registered durable queue exactly like +// `FLUSHALL` does (the registry has no way to scope to one db). Two +// different dbs sharing an MQ queue NAME is not a supported configuration. + +/// Tombstone any durable MQ stream(s) removed by a generic-keyspace +/// `DEL`/`UNLINK`. For each key argument found in +/// `s.durable_queue_registry`, removes the registry entry, deletes the +/// stream from `s.databases[db_index]` (idempotent if the generic command +/// already did so), and appends + replicates an `MqDrop` WAL tombstone. +pub(crate) fn auto_drop_mq_streams( + s: &mut crate::shard::slice::ShardSlice, + cmd_args: &[Frame], + db_index: usize, +) { + let Some(reg) = s.durable_queue_registry.as_mut() else { + return; + }; + if reg.is_empty() { + return; + } + let mut dropped: smallvec::SmallVec<[Bytes; 4]> = smallvec::SmallVec::new(); + for arg in cmd_args { + if let Frame::BulkString(key) = arg { + if reg.get(key).is_some() { + reg.remove(key); + dropped.push(key.clone()); + } + } + } + if dropped.is_empty() { + return; + } + emit_mq_drops(s, db_index, &dropped); +} + +/// Tombstone every durable MQ stream cleared by `FLUSHDB`/`FLUSHALL`. Both +/// clear the WHOLE registry (see the module-level scope note above) — the +/// only difference from `FLUSHALL` would be db-scoping, which the registry +/// cannot represent. +pub(crate) fn auto_drop_mq_streams_on_flush( + s: &mut crate::shard::slice::ShardSlice, + db_index: usize, +) { + let Some(reg) = s.durable_queue_registry.as_mut() else { + return; + }; + if reg.is_empty() { + return; + } + let dropped: smallvec::SmallVec<[Bytes; 4]> = reg.iter().map(|(k, _)| k.clone()).collect(); + for key in &dropped { + reg.remove(key); + } + emit_mq_drops(s, db_index, &dropped); +} + +/// Shared tail: append + replicate one `MqDrop` WAL record per already- +/// removed key. `db_index` is the record's db attribution — matches the +/// `MqCreate` convention of tagging the record with the acting command's +/// db, not a per-queue stored value (the registry has none). +fn emit_mq_drops(s: &mut crate::shard::slice::ShardSlice, db_index: usize, dropped: &[Bytes]) { + let shard_id = s.shard_id; + for key in dropped { + let payload = crate::mq::wal::encode_mq_drop(db_index as u32, key); + replicate_mq_record(shard_id, db_index, crate::mq::wal::MQ_REPL_DROP, &payload); + if let Some(ref tx) = s.wal_append_tx { + if tx + .try_send(( + crate::persistence::wal_v3::record::WalRecordType::MqDrop, + Bytes::from(payload), + )) + .is_err() + { + crate::command::info_reclamation::RECL_WAL_APPEND_CHANNEL_DROPPED_TOTAL + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tracing::error!( + "MqDrop WAL append channel rejected a tombstone record for a deleted \ + durable MQ stream AFTER the in-memory removal was applied — this \ + queue may resurrect on crash recovery. The channel is drained by \ + this same shard thread every 1ms (capacity 4096), so this indicates \ + a pathological single-tick burst." + ); + } + } + } +} + // ── Subcommand handlers ─────────────────────────────────────────────────────── /// MQ.CREATE — owner creates the durable stream + registry entry. diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs index 125bb1371..1e2635dd0 100644 --- a/src/shard/shared_databases.rs +++ b/src/shard/shared_databases.rs @@ -435,12 +435,13 @@ struct MqReplayStats { pop: u64, ack: u64, trigger: u64, + drop: u64, skipped: u64, } impl MqReplayStats { fn total(&self) -> u64 { - self.create + self.push + self.pop + self.ack + self.trigger + self.create + self.push + self.pop + self.ack + self.trigger + self.drop } } @@ -713,6 +714,31 @@ pub(crate) fn apply_mq_trigger( reg.register(bytes::Bytes::from(trig_key), entry); } +/// Apply a single decoded MqDrop record: tombstones a durable MQ stream +/// deleted via generic keyspace `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL` (kernel +/// M3 stage 3 / task #46). Removes the registry entry (by key — the +/// registry itself carries no db index, mirroring `apply_mq_create`'s +/// existing db-lax registry design) and deletes the stream from the +/// record's own db. Idempotent: a key with no registry entry / no stream is +/// a harmless no-op (e.g. replaying a Drop for a stream a prior malformed +/// record already failed to (re)create). +/// +/// # Ordering +/// +/// Applied strictly in WAL order alongside every other MQ record (task #46 +/// review): a `Drop` only undoes records that precede it for this key — a +/// LATER `MqCreate` for the same key re-materializes normally, so +/// create -> drop -> create survives a kill-9 with the second incarnation +/// intact. +pub(crate) fn apply_mq_drop(target: &mut T, db_index: usize, key: &[u8]) { + if let Some(reg) = target.mq_durable_queue_registry_mut().as_mut() { + reg.remove(key); + } + if let Some(db) = target.mq_databases_mut().get_mut(db_index) { + let _ = db.remove_counting_cold(key); + } +} + /// Decode + dispatch one MQ WAL record to its `apply_mq_*` handler, /// skip-and-warn on any decode failure (malformed payload OR an /// unsupported/future version byte). @@ -791,6 +817,13 @@ fn apply_mq_wal_record( } None => warn_skip_mq_record(shard_id, "MqTrigger", payload, stats), }, + WalRecordType::MqDrop => match crate::mq::wal::decode_mq_drop(payload) { + Some((db_index, key)) => { + apply_mq_drop(init, db_index as usize, &key); + stats.drop += 1; + } + None => warn_skip_mq_record(shard_id, "MqDrop", payload, stats), + }, _ => {} } } @@ -863,7 +896,8 @@ pub fn replay_mq_wal( | WalRecordType::MqAck | WalRecordType::MqPush | WalRecordType::MqPop - | WalRecordType::MqTrigger => { + | WalRecordType::MqTrigger + | WalRecordType::MqDrop => { handle_record(record.record_type, &record.payload); } WalRecordType::Command => { @@ -873,7 +907,8 @@ pub fn replay_mq_wal( | WalRecordType::MqAck | WalRecordType::MqPush | WalRecordType::MqPop - | WalRecordType::MqTrigger => { + | WalRecordType::MqTrigger + | WalRecordType::MqDrop => { handle_record(inner.record_type, &inner.payload); } _ => {} @@ -903,13 +938,14 @@ pub fn replay_mq_wal( if stats.total() > 0 || stats.skipped > 0 { tracing::info!( "Shard {}: replayed MQ WAL — {} create, {} push, {} pop, {} ack, \ - {} trigger record(s), {} skipped (malformed/unsupported version)", + {} trigger, {} drop record(s), {} skipped (malformed/unsupported version)", shard_id, stats.create, stats.push, stats.pop, stats.ack, stats.trigger, + stats.drop, stats.skipped, ); } @@ -1873,6 +1909,185 @@ mod tests { ); } + // ── MqDrop tombstone replay (kernel M3 stage 3 / task #46) ────────────── + + #[test] + fn test_replay_mq_wal_drop_tombstones_stream_and_registry() { + // Create + push, then Drop: the stream must be gone from the db AND + // the registry entry removed after replay. + use crate::mq::wal::{encode_mq_create, encode_mq_drop, encode_mq_push}; + use crate::persistence::wal_v3::record::WalRecordType; + + let tmp = tempfile::tempdir().expect("tempdir"); + let queue_key = b"dropped-queue".to_vec(); + + let dbs = vec![vec![Database::new()]]; + let (_shared, mut inits) = ShardDatabases::new(dbs); + + write_mq_wal_records( + tmp.path(), + 0, + &[ + (WalRecordType::MqCreate, encode_mq_create(0, &queue_key, 3)), + ( + WalRecordType::MqPush, + encode_mq_push( + 0, + &queue_key, + 1, + 0, + &[( + bytes::Bytes::from_static(b"f"), + bytes::Bytes::from_static(b"v"), + )], + ), + ), + (WalRecordType::MqDrop, encode_mq_drop(0, &queue_key)), + ], + ); + + replay_mq_wal(&mut inits, tmp.path()); + + assert!( + inits[0] + .durable_queue_registry + .as_ref() + .is_none_or(|reg| reg.get(&queue_key).is_none()), + "MqDrop must remove the registry entry" + ); + let db = inits[0].databases.get_mut(0).expect("db 0"); + assert!( + db.get_stream_mut(&queue_key).unwrap().is_none(), + "MqDrop must remove the stream from the db" + ); + } + + #[test] + fn test_replay_mq_wal_drop_only_kills_prior_records() { + // create -> drop -> create -> push must leave the SECOND + // incarnation intact: a Drop only undoes records that precede it in + // WAL order, never a later MqCreate/MqPush for the same key. + use crate::mq::wal::{encode_mq_create, encode_mq_drop, encode_mq_push}; + use crate::persistence::wal_v3::record::WalRecordType; + + let tmp = tempfile::tempdir().expect("tempdir"); + let queue_key = b"recreated-queue".to_vec(); + + let dbs = vec![vec![Database::new()]]; + let (_shared, mut inits) = ShardDatabases::new(dbs); + + write_mq_wal_records( + tmp.path(), + 0, + &[ + (WalRecordType::MqCreate, encode_mq_create(0, &queue_key, 3)), + ( + WalRecordType::MqPush, + encode_mq_push( + 0, + &queue_key, + 1, + 0, + &[( + bytes::Bytes::from_static(b"f"), + bytes::Bytes::from_static(b"old"), + )], + ), + ), + (WalRecordType::MqDrop, encode_mq_drop(0, &queue_key)), + (WalRecordType::MqCreate, encode_mq_create(0, &queue_key, 5)), + ( + WalRecordType::MqPush, + encode_mq_push( + 0, + &queue_key, + 2, + 0, + &[( + bytes::Bytes::from_static(b"f"), + bytes::Bytes::from_static(b"new"), + )], + ), + ), + ], + ); + + replay_mq_wal(&mut inits, tmp.path()); + + assert!( + inits[0] + .durable_queue_registry + .as_ref() + .and_then(|reg| reg.get(&queue_key)) + .is_some(), + "the SECOND MqCreate must re-register the queue" + ); + let db = inits[0].databases.get_mut(0).expect("db 0"); + let stream = db + .get_stream_mut(&queue_key) + .unwrap() + .expect("second incarnation stream must exist"); + assert_eq!( + stream.entries.len(), + 1, + "only the SECOND incarnation's push must survive — the first \ + incarnation's push preceded the Drop and must not resurrect" + ); + let entry = stream.entries.values().next().expect("exactly one entry"); + assert_eq!( + entry[0].1.as_ref(), + b"new", + "the surviving entry must be from the SECOND incarnation, not \ + a stale first-incarnation record the Drop should have killed" + ); + } + + #[test] + fn test_replay_mq_wal_drop_malformed_skipped_not_fatal() { + use crate::mq::wal::{encode_mq_create, encode_mq_push}; + use crate::persistence::wal_v3::record::WalRecordType; + + let tmp = tempfile::tempdir().expect("tempdir"); + let queue_key = b"q".to_vec(); + + let dbs = vec![vec![Database::new()]]; + let (_shared, mut inits) = ShardDatabases::new(dbs); + + let bogus_drop = vec![99u8, 0, 0, 0, 0]; // bad version byte + write_mq_wal_records( + tmp.path(), + 0, + &[ + (WalRecordType::MqCreate, encode_mq_create(0, &queue_key, 3)), + (WalRecordType::MqDrop, bogus_drop), + ( + WalRecordType::MqPush, + encode_mq_push( + 0, + &queue_key, + 1, + 0, + &[( + bytes::Bytes::from_static(b"f"), + bytes::Bytes::from_static(b"v"), + )], + ), + ), + ], + ); + + // Must not panic; the malformed Drop is skipped, the stream must + // survive with the later well-formed Push applied. + replay_mq_wal(&mut inits, tmp.path()); + + let db = inits[0].databases.get_mut(0).expect("db 0"); + assert_eq!( + db.get_or_create_stream(&queue_key).unwrap().entries.len(), + 1, + "a well-formed record after a skipped malformed MqDrop must still apply" + ); + } + #[test] fn test_replay_mq_wal_missing_wal_v3_subdir_is_a_noop() { // Regression guard for bug (a): files sitting directly under diff --git a/tests/crash_matrix_cross_plane/tests_seeded_red.rs b/tests/crash_matrix_cross_plane/tests_seeded_red.rs index e2f91511a..857674867 100644 --- a/tests/crash_matrix_cross_plane/tests_seeded_red.rs +++ b/tests/crash_matrix_cross_plane/tests_seeded_red.rs @@ -12,7 +12,19 @@ //! already contain the queue name, so waiting for that substring after a //! `DEL` would false-positive-match the EARLIER create/push records. A //! later, unrelated marker proves (by WAL append ordering) that everything -//! before it — including the delete — reached disk. +//! before it — including the delete — reached disk. As of task #46, DEL/ +//! FLUSHALL's `MqDrop` tombstone lands on the wal-v3 MQ plane (its own +//! fire-and-forget channel, drained on a separate 1ms tick from the AOF +//! writer), so the follow-up sync marker must ALSO be a wal-v3-family write +//! (a throwaway durable queue's own `MQ.PUSH`) — an AOF-only marker proves +//! nothing about whether the MqDrop record itself reached disk. +//! +//! GREEN as of kernel M3 stage 3 (task #46): `MqDrop` (WAL discriminant +//! 0x75, `src/mq/wal.rs`) tombstones a durable stream deleted via generic +//! `DEL`/`UNLINK`/`FLUSHDB`/`FLUSHALL`, applied strictly in WAL order by +//! `replay_mq_wal`/`apply_mq_drop` (`src/shard/shared_databases.rs`) — the +//! former RED cell below (`cross_plane_seeded_red_mq_generic_del_resurrection`, +//! now un-gated) and its FLUSHALL sibling both pass. use crate::harness::{self, Config}; use crate::planes::*; @@ -21,23 +33,14 @@ use crate::resp::Conn; /// MQ: `DEL ` (generic keyspace delete, NOT an `MQ.*` command) must /// tombstone the stream so it does not resurrect on the next restart. /// -/// RED (expected, per PR #291's changelog and brief §1.4): `replay_mq_wal` -/// has no way to represent "this queue was deleted after these pushes" — -/// there is no MqDelete/tombstone WAL record — so replay re-materializes -/// the full pre-delete stream content regardless of the `DEL`. +/// Formerly RED (PR #291's changelog and brief §1.4): `replay_mq_wal` had +/// no way to represent "this queue was deleted after these pushes" — there +/// was no MqDrop tombstone WAL record — so replay re-materialized the full +/// pre-delete stream content regardless of the `DEL`. Fixed by the `MqDrop` +/// record (task #46); un-gated (no more `harness::red_guard`). #[test] #[ignore] // Requires built release binary; run explicitly. fn cross_plane_seeded_red_mq_generic_del_resurrection() { - if !harness::red_guard( - "MQ streams deleted via generic DEL are not tombstoned and \ - resurrect with full content on the next restart — no MQ tombstone \ - WAL record exists yet (PR #291 changelog, kernel M3 brief §1.4). \ - Same bug class as the already-fixed KV/vector cold-plane \ - resurrection (PR #257), not yet applied to MQ. Tracked \ - separately — not this stage's job to fix.", - ) { - return; - } let cfg = Config::PROD_S1; let dir = harness::unique_dir("seededred-mq"); let (guard, port) = harness::spawn_moon_on(&dir, &cfg, &[]); @@ -65,14 +68,26 @@ fn cross_plane_seeded_red_mq_generic_del_resurrection() { ); // Sync a LATER, unrelated write so its position proves the DEL (which - // precedes it in append order) is also durable. `DEL` is a generic - // keyspace command — same log family as any other keyspace write, i.e. - // the AOF, not wal-v3 (confirmed empirically: plain commands never - // appear in wal-v3 at all) — so the follow-up sync marker must be - // AOF-family too, or it proves nothing about the DEL's own durability. + // precedes it in append order) is also durable. `DEL` itself logs to + // the AOF (generic keyspace command family), but as of task #46 it ALSO + // emits an `MqDrop` record on the wal-v3 MQ plane (a separate + // fire-and-forget channel, drained on its own 1ms tick) — an AOF-only + // sync marker proves the AOF is flushed but says nothing about whether + // the wal-v3 MqDrop record reached disk before the crash. The follow-up + // sync marker must therefore be wal-v3-family too: a throwaway durable + // queue's own MQ.PUSH, appended (and hence flushed) strictly AFTER the + // DEL's MqDrop record in this shard's wal-v3 file. + let sync_q = "seededred-mq-post-del-sync-queue"; + mq_create(&mut c, sync_q, 0); let marker2 = "SEEDEDRED-MQ-POST-DEL-SYNC".to_string(); - kv_set(&mut c, "seededred-mq-post-del-marker", &marker2); - harness::wait_for_aof_bytes_any_shard(&dir, marker2.as_bytes()); + mq_push(&mut c, sync_q, "final", &marker2); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker2.as_bytes()); + + // Keep the pre-existing AOF-family sync too — DEL's own AOF entry is + // still a durability requirement independent of the MqDrop tombstone. + let marker2_aof = "SEEDEDRED-MQ-POST-DEL-AOF-SYNC".to_string(); + kv_set(&mut c, "seededred-mq-post-del-marker", &marker2_aof); + harness::wait_for_aof_bytes_any_shard(&dir, marker2_aof.as_bytes()); harness::crash(guard, port); let (guard2, port2) = harness::spawn_moon_on(&dir, &cfg, &[]); @@ -87,6 +102,110 @@ fn cross_plane_seeded_red_mq_generic_del_resurrection() { drop(guard2); } +/// MQ: `FLUSHALL` must tombstone every durable stream it cleared, the same +/// as a targeted `DEL` above — `auto_drop_mq_streams_on_flush` +/// (`src/shard/mq_exec.rs`) is the FLUSHALL/FLUSHDB sibling of the DEL/ +/// UNLINK hook proven by `cross_plane_seeded_red_mq_generic_del_resurrection`. +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_mq_flushall_resurrection() { + let cfg = Config::PROD_S1; + let dir = harness::unique_dir("mq-flushall"); + let (guard, port) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c = Conn::open(port); + let q = "mq-flushall-queue"; + + mq_create(&mut c, q, 0); + mq_push(&mut c, q, "seq", "1"); + let marker1 = "MQ-FLUSHALL-PRE-SYNC".to_string(); + mq_push(&mut c, q, "final", &marker1); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker1.as_bytes()); + + assert_eq!(xlen(&mut c, q), 2, "sanity: 2 messages before FLUSHALL"); + match c.cmd_s(&["FLUSHALL"]) { + crate::resp::Resp::Simple(_) => {} + other => panic!("FLUSHALL failed: {other:?}"), + } + assert_eq!( + xlen(&mut c, q), + 0, + "sanity: FLUSHALL cleared it immediately" + ); + + // Same wal-v3-family sync requirement as the DEL test above: FLUSHALL's + // `MqDrop` tombstones land on the wal-v3 MQ plane, not the AOF, so the + // proof-of-durability marker must be a wal-v3 write too. A NEW queue + // created after the FLUSHALL (FLUSHALL clears definitions' contents but + // this is a fresh key) whose own MQ.PUSH is appended strictly after + // every tombstone this FLUSHALL emitted. + let sync_q = "mq-flushall-post-sync-queue"; + mq_create(&mut c, sync_q, 0); + let marker2 = "MQ-FLUSHALL-POST-SYNC".to_string(); + mq_push(&mut c, sync_q, "final", &marker2); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker2.as_bytes()); + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c2 = Conn::open(port2); + assert_eq!( + xlen(&mut c2, q), + 0, + "MQ stream {q:?} must NOT resurrect after FLUSHALL + kill-9" + ); + drop(guard2); +} + +/// MQ: create -> drop -> create -> kill-9 must survive with the SECOND +/// incarnation intact — an `MqDrop` tombstone only kills the records that +/// precede it for a key, never a later `MqCreate`/`MqPush` for the same +/// key (replay is strictly WAL-ordered; see `apply_mq_drop`'s doc comment). +#[test] +#[ignore] // Requires built release binary; run explicitly. +fn cross_plane_mq_create_drop_create_survives() { + let cfg = Config::PROD_S1; + let dir = harness::unique_dir("mq-recreate"); + let (guard, port) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c = Conn::open(port); + let q = "mq-recreate-queue"; + + // First incarnation: created, pushed, then deleted. + mq_create(&mut c, q, 0); + mq_push(&mut c, q, "seq", "1"); + c.cmd_s(&["DEL", q]); + assert_eq!( + xlen(&mut c, q), + 0, + "sanity: DEL removed the first incarnation" + ); + + // Second incarnation: re-created with DIFFERENT content — must survive + // the crash intact, not be killed by the first incarnation's tombstone. + mq_create(&mut c, q, 0); + mq_push(&mut c, q, "seq", "100"); + mq_push(&mut c, q, "seq", "101"); + let marker = "MQ-RECREATE-POST-SYNC".to_string(); + mq_push(&mut c, q, "final", &marker); + harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker.as_bytes()); + assert_eq!( + xlen(&mut c, q), + 3, + "sanity: 3 messages in second incarnation" + ); + + harness::crash(guard, port); + + let (guard2, port2) = harness::spawn_moon_on(&dir, &cfg, &[]); + let mut c2 = Conn::open(port2); + assert_eq!( + xlen(&mut c2, q), + 3, + "second incarnation of {q:?} must survive kill-9 with all 3 \ + messages — the first incarnation's MqDrop tombstone must not \ + kill records that come AFTER it in WAL order" + ); + drop(guard2); +} + /// WS analogue: `WS DROP ` (the workspace registry's OWN delete /// operation — `WorkspaceRegistry` is process-global, not a normal keyspace /// key, so a *generic* `DEL` does not apply to it the way it does to an MQ