feat(replication): MQ plane — live stream + PSYNC registry blob + RDB Stream codec data-loss fix (Wave B stage 2b, task #34)

.github/workflows/fuzz.yml

@@ -37,6 +37,8 @@ jobs:
           - fts_query_parse
           - mq_wal_record
           - ws_registry_record
+          - mq_registry_blob
+          - redis_rdb_load
     steps:
       - uses: actions/checkout@v7
       - uses: dtolnay/rust-toolchain@nightly
@@ -90,6 +92,8 @@ jobs:
           - fts_query_parse
           - mq_wal_record
           - ws_registry_record
+          - mq_registry_blob
+          - redis_rdb_load
     steps:
       - uses: actions/checkout@v7
       - uses: dtolnay/rust-toolchain@nightly

CHANGELOG.md

@@ -72,16 +72,67 @@ on the multi-shard master path) is installed authoritatively at
 `load_snapshot`, same convention as the graph/vector/text aux blobs. New
 `ws_registry_record` cargo-fuzz target covers the blob decoder.
 
-Split the former combined `warn_unreplicated_plane` fail-loud marker
-(`handler_monoio/ft.rs`) into per-plane functions — the WS half is retired
-now that this plane replicates; `warn_mq_unreplicated` survives for MQ,
-which is not yet wired in.
+The former combined `warn_unreplicated_plane` fail-loud marker
+(`handler_monoio/ft.rs`) is fully retired: its WS half by this WS-plane
+work, and its MQ half by the MQ-plane replication entry below.
 
 New `tests/replication_ws.rs`: live-stream parity (id + `created_at`
 round-trip through `WS.INFO`/`WS.AUTH` on the replica), snapshot-leg
 backfill, `WS.DROP` propagation, and a `--shards 4` leg exercising the
 shard-0 hop from connections that may land on any shard.
 
+### Added — MQ-plane replication (Wave B stage 2b)
+
+Durable-queue MQ.* mutations now replicate to attached replicas, closing the
+plane gap `warn_unreplicated_plane` previously fail-loud-warned about for
+MQ.*. Builds on the stage-2a MQ WAL effect records (PR #291):
+
+- **Live stream**: `shard::mq_exec::replicate_mq_record` emits the SAME
+  encoded `MqCreate`/`MqPush`/`MqPop`/`MqAck`/`MqTrigger` payload bytes
+  already built for the WAL as one of five synthetic `MQ._REPL.*`
+  pseudo-commands (never dispatched to real clients), gated single-shard-only
+  (`num_shards == 1`) — the same posture graph's own live path takes for its
+  multi-shard gap. `MQ.PUBLISH`'s TXN-materialization leg
+  (`server::conn::handler_monoio::txn.rs`) replicates the same way at
+  commit. `replication::apply::apply_mq` decodes and applies through the
+  SAME `apply_mq_*` engine boot-time WAL replay uses
+  (`shard::shared_databases`, generalized over a new `MqApplyTarget` trait so
+  `ShardSliceInit` boot replay and `ShardSlice` live apply share one codec
+  instead of two).
+- **FULLRESYNC**: a per-shard `MOON_AUX_MQ_REGISTRY` aux blob
+  (`replication::mq_sync`) carries the durable-queue registry and trigger
+  registry — the shard-level bookkeeping that lives outside the keyspace.
+  Installed additively into every replica shard
+  (`mq_sync::install_mq_registry_many`), mirroring
+  `graph_sync::install_graph_store_many`.
+- **Fixed a pre-existing FULLRESYNC data-loss bug found along the way**: the
+  PSYNC RDB codec (`persistence::redis_rdb`, distinct from the
+  `persistence::rdb` SAVE/BGSAVE codec) serialized `Stream` values as a
+  placeholder `"__stream__:<len>"` string, discarding all entries/PEL/
+  consumer-group state. A full `RDB_TYPE_STREAM_MOON` (0xC8) codec — ported
+  from `persistence::rdb`'s battle-tested Stream serializer — now carries
+  entries, `last_id`, consumer groups (PEL + per-consumer pending), the
+  `durable` flag, and `max_delivery_count`. Without this fix, MQ FULLRESYNC
+  backfill would silently lose every durable queue's message content.
+  **Operator note — upgrade masters and replicas together** if any durable
+  Streams exist: a pre-fix replica syncing from a fixed master hard-fails
+  FULLRESYNC on the new `RDB_TYPE_STREAM_MOON` tag (and retries on its
+  reconnect loop until upgraded), while a fixed replica syncing from a
+  pre-fix master absorbs the old placeholder as a corrupted string value —
+  both inherent to fixing a wire-format bug, neither loses data already
+  durable in the master's WAL.
+- Two new `cargo-fuzz` targets: `mq_registry_blob` (the new
+  `install_mq_registry_many` decoder) and `redis_rdb_load` (the FULLRESYNC
+  RDB loader as a whole, including the new Stream type tag) — wired into
+  both PR and nightly fuzz matrices.
+- `tests/replication_mq.rs`: live-stream, snapshot-backfill, and a pinned
+  multi-shard-live-write-not-streamed limitation test (shards=1 scope,
+  matching graph's own precedent).
+
+A multi-shard master still does not live-stream MQ writes (durability is
+unaffected — WAL + FULLRESYNC still cover it); tracked as a known follow-up
+alongside graph's own multi-shard live-stream gap.
+
 ### Fixed — docs site strict build red since 2026-07-08 (root-relative links)
 
 `docs/guides/tuning.md` (PR #245) and `docs/PRODUCTION-CONTRACT.md` (PR #263)

fuzz/Cargo.toml

@@ -93,3 +93,13 @@ doc = false
 name = "ws_registry_record"
 path = "fuzz_targets/ws_registry_record.rs"
 doc = false
+
+[[bin]]
+name = "mq_registry_blob"
+path = "fuzz_targets/mq_registry_blob.rs"
+doc = false
+
+[[bin]]
+name = "redis_rdb_load"
+path = "fuzz_targets/redis_rdb_load.rs"
+doc = false

fuzz/fuzz_targets/mq_registry_blob.rs

@@ -0,0 +1,69 @@
+#![no_main]
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+
+use libfuzzer_sys::fuzz_target;
+use moon::replication::mq_sync::install_mq_registry_many;
+use moon::shard::shared_databases::ShardStoreMemory;
+use moon::shard::slice::{ShardSlice, ShardSliceInit, init_shard, with_shard};
+use moon::storage::db::Database;
+use moon::text::store::TextStore;
+use moon::transaction::{DeferredHnswInserts, KvWriteIntents};
+use moon::vector::store::VectorStore;
+
+/// Fuzz the Wave B stage 2b MQ-registry FULLRESYNC blob decoder
+/// (`replication::mq_sync::install_mq_registry_many`), mirroring
+/// `mq_wal_record.rs`'s coverage of the sibling WAL op-blob decoders.
+///
+/// `data` arrives on the replica exactly like `apply::load_snapshot` feeds
+/// it: bytes read straight off a `MOON_AUX_MQ_REGISTRY` RDB aux field sent
+/// by a master over the wire — attacker/corruption-controlled (a malicious
+/// or buggy master, a torn transfer, a version skew). The decoder
+/// (`mq_sync::Cursor` + `install_one`) is documented to return `None` on
+/// ANY malformed input, truncated blob, or unknown version byte — never
+/// panic, never read out of bounds, never allocate unboundedly off an
+/// attacker length prefix (`len_checked` bounds every prefix against the
+/// remaining blob before `Bytes::copy_from_slice`).
+///
+/// `ShardSlice` state lives in a `thread_local!`, so — mirroring
+/// `mq_sync.rs`'s own unit tests — each fuzz iteration installs into a
+/// fresh `ShardSlice` on a throwaway OS thread.
+fuzz_target!(|data: &[u8]| {
+    let data = data.to_vec();
+    let _ = std::thread::spawn(move || {
+        init_shard(make_fuzz_slice());
+        with_shard(|s: &mut ShardSlice| {
+            // Should not panic regardless of input.
+            let _ = install_mq_registry_many(s, &[data]);
+        });
+    })
+    .join();
+});
+
+fn make_fuzz_slice() -> ShardSlice {
+    let databases: Box<[Database]> = (0..1).map(|_| Database::new()).collect();
+    ShardSlice::new(ShardSliceInit {
+        shard_id: 0,
+        databases,
+        vector_store: VectorStore::new(),
+        text_store: TextStore::new(),
+        // moon-fuzz always builds `moon` with the "graph" feature (see
+        // fuzz/Cargo.toml), so this field is unconditional here even
+        // though it's `#[cfg(feature = "graph")]` inside moon itself.
+        graph_store: moon::graph::store::GraphStore::new(),
+        kv_write_intents: KvWriteIntents::new(),
+        deferred_hnsw_inserts: DeferredHnswInserts::new(),
+        temporal_registry: None,
+        temporal_kv_index: None,
+        durable_queue_registry: None,
+        trigger_registry: None,
+        wal_append_tx: None,
+        estimated_memory: Arc::new(AtomicUsize::new(0)),
+        store_memory: Arc::new(ShardStoreMemory {
+            vector: AtomicUsize::new(0),
+            text: AtomicUsize::new(0),
+            graph: AtomicUsize::new(0),
+            lua: AtomicUsize::new(0),
+        }),
+    })
+}

fuzz/fuzz_targets/redis_rdb_load.rs

@@ -0,0 +1,22 @@
+#![no_main]
+use libfuzzer_sys::fuzz_target;
+
+/// Fuzz the Redis-compatible RDB loader used by PSYNC FULLRESYNC
+/// (`persistence::redis_rdb::load_rdb`), distinct from the SAVE/BGSAVE
+/// codec covered by `rdb_load.rs` (`persistence::rdb::load`).
+///
+/// This is the codec `replication::master` streams to a replica and
+/// `replication::apply::load_snapshot` decodes on the replica side; a
+/// malicious or corrupted master (or a bit-flipped wire transfer) must
+/// never panic the replica. Exercises magic/version validation, type-tag
+/// dispatch (including the private `RDB_TYPE_STREAM_MOON` extension added
+/// for Wave B stage 2b MQ replication), length-prefixed collection bounds
+/// checks (`check_alloc_bound`), and CRC/EOF handling — operates directly
+/// on the in-memory byte slice (no temp file needed, unlike `rdb_load.rs`).
+fuzz_target!(|data: &[u8]| {
+    let mut databases: Vec<moon::storage::db::Database> =
+        (0..1).map(|_| moon::storage::db::Database::new()).collect();
+
+    // Should not panic regardless of input.
+    let _ = moon::persistence::redis_rdb::load_rdb(&mut databases, data);
+});

src/command/mq.rs

@@ -96,12 +96,12 @@ pub fn parse_mq_subcommand(args: &[Frame]) -> Result<&[u8], Frame> {
 ///
 /// `MQ` is registered blanket-WRITE in `command::metadata::COMMAND_META`
 /// (CREATE/PUSH/POP/ACK/TRIGGER/PUBLISH all mutate durable queue state), but
-/// DLQLEN only queries the dead-letter depth (same condition
-/// `try_handle_mq_command` already uses to skip `warn_unreplicated_plane`).
-/// A plain `MQ LEN` is referenced in that same skip-list but has no handler
-/// arm in `try_handle_mq_command` today (falls through to
-/// `ERR_MQ_UNKNOWN_SUB`) — it is included here defensively so that if it is
-/// ever wired up as a read, this classifier does not need a second change.
+/// DLQLEN only queries the dead-letter depth (the retired
+/// `warn_unreplicated_plane` marker skipped it for the same reason).
+/// A plain `MQ LEN` has no handler arm in `try_handle_mq_command` today
+/// (falls through to `ERR_MQ_UNKNOWN_SUB`) — it is included here
+/// defensively so that if it is ever wired up as a read, this classifier
+/// does not need a second change.
 /// Any subcommand not in this allow-list (including unknown ones) is
 /// conservatively treated as a write so `try_enforce_readonly` never
 /// false-negatives.

src/mq/trigger.rs

@@ -63,6 +63,13 @@ impl TriggerRegistry {
         self.entries.get_mut(key)
     }
 
+    /// Iterate over all entries (composite key, entry). Used by
+    /// `replication::mq_sync::export_mq_registry` (Wave B stage 2b) to
+    /// snapshot every registered trigger for FULLRESYNC.
+    pub fn iter(&self) -> impl Iterator<Item = (&Bytes, &TriggerEntry)> {
+        self.entries.iter()
+    }
+
     /// Return keys of entries whose pending fire time has elapsed.
     ///
     /// An entry is ready to fire when `pending_fire_ms > 0` and
@@ -306,4 +313,20 @@ mod tests {
 
         assert_eq!(reg.get(b"ws1:q1").unwrap().debounce_ms, 2000);
     }
+
+    #[test]
+    fn test_iter() {
+        let mut reg = TriggerRegistry::new();
+        reg.register(
+            Bytes::from_static(b"ws1:q1"),
+            make_entry(b"q1", b"CMD1", 100),
+        );
+        reg.register(
+            Bytes::from_static(b"ws1:q2"),
+            make_entry(b"q2", b"CMD2", 200),
+        );
+        let mut keys: Vec<&[u8]> = reg.iter().map(|(k, _)| k.as_ref()).collect();
+        keys.sort();
+        assert_eq!(keys, vec![&b"ws1:q1"[..], &b"ws1:q2"[..]]);
+    }
 }

src/mq/wal.rs

@@ -400,10 +400,63 @@ pub fn decode_mq_trigger(payload: &[u8]) -> Option<(Vec<u8>, Vec<u8>, Vec<u8>, u
     Some((trig_key, queue_key, callback_cmd, debounce_ms))
 }
 
+// ── Replication wire framing (Wave B stage 2b) ────────────────────────────
+//
+// MQ effect records are versioned BINARY payloads, not RESP commands (unlike
+// GRAPH.*/TEMPORAL.INVALIDATE-AT, which stream deterministic user-syntax
+// commands). Replication wraps each one in a synthetic pseudo-command that
+// carries the SAME `encode_mq_*` payload verbatim as a single bulk-string
+// arg: `MQ._REPL.PUSH <payload bytes>`. These names are NEVER dispatched to
+// a real client (no `CommandMeta`/ACL entry, no overlap with
+// `is_mq_command`'s `"MQ"` check) — they are matched only by
+// `replication::apply::apply_local`'s command-name routing, exactly like
+// `GraphReplayCollector::is_graph_command`, and decoded with the SAME
+// `decode_mq_*` functions above so the replica's apply arm and boot-time WAL
+// replay share one codec.
+
+/// Synthetic replication pseudo-command for [`WalRecordType::MqCreate`].
+pub const MQ_REPL_CREATE: &[u8] = b"MQ._REPL.CREATE";
+/// Synthetic replication pseudo-command for [`WalRecordType::MqPush`].
+pub const MQ_REPL_PUSH: &[u8] = b"MQ._REPL.PUSH";
+/// Synthetic replication pseudo-command for [`WalRecordType::MqPop`].
+pub const MQ_REPL_POP: &[u8] = b"MQ._REPL.POP";
+/// Synthetic replication pseudo-command for [`WalRecordType::MqAck`].
+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";
+
+/// True for any `MQ._REPL.*` synthetic replication pseudo-command emitted by
+/// [`crate::shard::mq_exec`]'s live fan-out. Used by
+/// `replication::apply::apply_local` to route a replicated command to the
+/// MQ apply engine instead of generic dispatch.
+///
+/// [`WalRecordType`]: crate::persistence::wal_v3::record::WalRecordType
+#[inline]
+pub fn is_mq_replay_command(cmd: &[u8]) -> bool {
+    cmd.eq_ignore_ascii_case(MQ_REPL_CREATE)
+        || cmd.eq_ignore_ascii_case(MQ_REPL_PUSH)
+        || cmd.eq_ignore_ascii_case(MQ_REPL_POP)
+        || cmd.eq_ignore_ascii_case(MQ_REPL_ACK)
+        || cmd.eq_ignore_ascii_case(MQ_REPL_TRIGGER)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_is_mq_replay_command() {
+        assert!(is_mq_replay_command(b"MQ._REPL.CREATE"));
+        assert!(is_mq_replay_command(b"mq._repl.push"));
+        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"));
+        assert!(!is_mq_replay_command(b"MQ.PUSH"));
+        assert!(!is_mq_replay_command(b"GRAPH.ADDNODE"));
+        assert!(!is_mq_replay_command(b""));
+    }
+
     // --- MqCreate roundtrip ---
 
     #[test]