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