fix(persistence): MqDrop tombstone — DEL/UNLINK/FLUSHALL/FLUSHDB no longer resurrect durable MQ streams on kill-9 (task #46)

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.
 

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);
 });

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<u8>, Vec<u8>, Vec<u8>, 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<u8> {
+    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<u8>)> {
+    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]

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());
         }

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;
             }

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(_) => {}

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);
     }
 }