From 75aa6f91aa820fc703e63670714fc6806075668b Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:00:23 +0700
Subject: [PATCH 01/10] fix(persistence): PSYNC RDB Stream codec discarded all
 content (data loss)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The FULLRESYNC RDB codec (persistence::redis_rdb, distinct from the
persistence::rdb SAVE/BGSAVE codec) serialized Stream values as a
placeholder "__stream__:<len>" string — entries, last_id, consumer
groups, PEL state, durable flag, and max_delivery_count were all
silently discarded on every replica full resync. A durable MQ queue
snapshotted this way would come back empty on the replica.

Add a full RDB_TYPE_STREAM_MOON (0xC8) codec, ported from
persistence::rdb's battle-tested Stream serializer: entries, last_id,
consumer groups (PEL + per-consumer pending), durable flag, and
max_delivery_count round-trip losslessly. Every new count field is
bounds-checked via check_alloc_bound before allocating, matching the
existing DoS-avoidance discipline in this module.

Also adds the MOON_AUX_MQ_REGISTRY aux-field constant used by the
Wave B stage 2b MQ replication work landing in subsequent commits.

author: Tin Dang
---
 src/persistence/redis_rdb.rs | 337 ++++++++++++++++++++++++++++++++++-
 1 file changed, 331 insertions(+), 6 deletions(-)

diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs
index 3b153bd7..1e78ca5a 100644
--- a/src/persistence/redis_rdb.rs
+++ b/src/persistence/redis_rdb.rs
@@ -20,6 +20,9 @@ use ordered_float::OrderedFloat;
 use crate::storage::compact_value::RedisValueRef;
 use crate::storage::db::Database;
 use crate::storage::entry::{Entry, RedisValue, current_time_ms};
+use crate::storage::stream::{
+    Consumer, ConsumerGroup, PendingEntry, Stream as StreamData, StreamId,
+};
 
 // ---------------------------------------------------------------------------
 // Redis RDB opcodes and type tags
@@ -39,6 +42,14 @@ const RDB_TYPE_LIST: u8 = 1;
 const RDB_TYPE_SET: u8 = 2;
 const RDB_TYPE_HASH: u8 = 4;
 const RDB_TYPE_ZSET_2: u8 = 5;
+/// Moon-private extension type tag (real Redis reserves 6-18 and uses 19+ for
+/// native stream listpack encodings; this format is consumed ONLY by moon's
+/// own replica — see module docs — so a private tag well outside Redis's
+/// range is safe and does not need to match Redis's on-wire stream format).
+/// Wave B stage 2b: replaces the pre-existing `"__stream__:<len>"` placeholder
+/// STRING encoding, which discarded all entry/PEL/consumer-group content and
+/// would type-corrupt a Stream key into a plain string on FULLRESYNC.
+const RDB_TYPE_STREAM_MOON: u8 = 200;
 
 // Integer-encoded string prefixes (11xxxxxx)
 const RDB_ENC_INT8: u8 = 0;
@@ -404,13 +415,62 @@ fn write_rdb_entry(buf: &mut Vec<u8>, key: &[u8], entry: &Entry, base_ts: u32) {
             }
         }
         RedisValueRef::Stream(stream) => {
-            // Streams: serialize as JSON string with type 0 for simplicity.
-            // Redis uses complex type 19+ for native stream encoding, but for
-            // PSYNC2 to our own replicas this is sufficient.
-            buf.push(RDB_TYPE_STRING);
+            // Full-fidelity moon-private stream encoding (Wave B stage 2b) —
+            // entries, last_id, consumer groups (PEL + per-consumer pending
+            // sets), and the MQ durability flags (`durable` /
+            // `max_delivery_count`), mirroring `persistence::rdb.rs`'s
+            // battle-tested Stream codec (that module's format is the same
+            // shape; kept as a separate encoder here because this module's
+            // string/length framing — `write_redis_string`/`write_length` —
+            // differs from that module's raw-LE-prefixed helpers).
+            buf.push(RDB_TYPE_STREAM_MOON);
             write_redis_string(buf, key);
-            let json = format!("__stream__:{}", stream.length);
-            write_redis_string(buf, json.as_bytes());
+            write_length(buf, stream.entries.len() as u64);
+            buf.extend_from_slice(&stream.last_id.ms.to_le_bytes());
+            buf.extend_from_slice(&stream.last_id.seq.to_le_bytes());
+            for (id, fields) in &stream.entries {
+                buf.extend_from_slice(&id.ms.to_le_bytes());
+                buf.extend_from_slice(&id.seq.to_le_bytes());
+                write_length(buf, fields.len() as u64);
+                for (field, value) in fields {
+                    write_redis_string(buf, field);
+                    write_redis_string(buf, value);
+                }
+            }
+            write_length(buf, stream.groups.len() as u64);
+            for (group_name, group) in &stream.groups {
+                write_redis_string(buf, group_name);
+                buf.extend_from_slice(&group.last_delivered_id.ms.to_le_bytes());
+                buf.extend_from_slice(&group.last_delivered_id.seq.to_le_bytes());
+                write_length(buf, group.pel.len() as u64);
+                for (id, pe) in &group.pel {
+                    buf.extend_from_slice(&id.ms.to_le_bytes());
+                    buf.extend_from_slice(&id.seq.to_le_bytes());
+                    write_redis_string(buf, &pe.consumer);
+                    buf.extend_from_slice(&pe.delivery_time.to_le_bytes());
+                    buf.extend_from_slice(&pe.delivery_count.to_le_bytes());
+                }
+                write_length(buf, group.consumers.len() as u64);
+                for (cname, consumer) in &group.consumers {
+                    write_redis_string(buf, cname);
+                    buf.extend_from_slice(&consumer.seen_time.to_le_bytes());
+                    write_length(buf, consumer.pending.len() as u64);
+                    for (id, _) in &consumer.pending {
+                        buf.extend_from_slice(&id.ms.to_le_bytes());
+                        buf.extend_from_slice(&id.seq.to_le_bytes());
+                    }
+                }
+            }
+            // MQ durability flags: a plain XADD stream round-trips these as
+            // `false`/`0` (harmless); an MQ.CREATE-managed stream needs them
+            // to survive FULLRESYNC so MQ.POP/MQ.ACK keep working on the
+            // freshly-synced replica without waiting for a registry-driven
+            // re-CREATE (the registry blob — `MOON_AUX_MQ_REGISTRY` — carries
+            // the SAME `max_delivery_count`, redundantly but harmlessly, for
+            // shard-level bookkeeping independent of which db the stream
+            // lives in).
+            buf.push(u8::from(stream.durable));
+            buf.extend_from_slice(&stream.max_delivery_count.to_le_bytes());
         }
     }
 }
@@ -455,6 +515,13 @@ pub const MOON_AUX_GRAPH_STORE: &[u8] = b"moon-graph-store";
 /// the merged RDB regardless of `--shards` (shard-0-authoritative; see
 /// `ws_sync` module docs). See [`MOON_AUX_VECTOR_DEFS`].
 pub const MOON_AUX_WORKSPACE_REGISTRY: &[u8] = b"moon-ws-registry";
+/// Moon replication aux key: one shard's MQ durable-queue + trigger registry
+/// snapshot (`replication::mq_sync::export_mq_registry` blob — see that
+/// module for the format). Stream CONTENT itself rides the ordinary keyspace
+/// RDB body (via `RDB_TYPE_STREAM_MOON`); this aux blob carries only the
+/// shard-level metadata that lives OUTSIDE the keyspace (`ShardSlice`'s
+/// `durable_queue_registry` / `trigger_registry`). See [`MOON_AUX_VECTOR_DEFS`].
+pub const MOON_AUX_MQ_REGISTRY: &[u8] = b"moon-mq-registry";
 
 /// `write_rdb_refs` plus moon-private AUX fields written immediately after
 /// the standard header aux block (before any SELECTDB), which is what lets
@@ -783,6 +850,132 @@ fn read_rdb_entry(
             }
             entry
         }
+        RDB_TYPE_STREAM_MOON => {
+            let (entry_count, _) = read_length(cursor)?;
+            check_alloc_bound(cursor, entry_count, 1, "stream_entries")?;
+            let mut last_id_ms = [0u8; 8];
+            let mut last_id_seq = [0u8; 8];
+            cursor.read_exact(&mut last_id_ms)?;
+            cursor.read_exact(&mut last_id_seq)?;
+            let mut stream = StreamData::new();
+            stream.last_id = StreamId {
+                ms: u64::from_le_bytes(last_id_ms),
+                seq: u64::from_le_bytes(last_id_seq),
+            };
+            for _ in 0..entry_count {
+                let mut ms_buf = [0u8; 8];
+                let mut seq_buf = [0u8; 8];
+                cursor.read_exact(&mut ms_buf)?;
+                cursor.read_exact(&mut seq_buf)?;
+                let id = StreamId {
+                    ms: u64::from_le_bytes(ms_buf),
+                    seq: u64::from_le_bytes(seq_buf),
+                };
+                let (field_count, _) = read_length(cursor)?;
+                check_alloc_bound(cursor, field_count, 1, "stream_fields")?;
+                let mut fields = Vec::with_capacity(field_count as usize);
+                for _ in 0..field_count {
+                    let field = read_redis_string(cursor)?;
+                    let value = read_redis_string(cursor)?;
+                    fields.push((Bytes::from(field), Bytes::from(value)));
+                }
+                stream.entries.insert(id, fields);
+                stream.length += 1;
+            }
+
+            let (group_count, _) = read_length(cursor)?;
+            check_alloc_bound(cursor, group_count, 1, "stream_groups")?;
+            for _ in 0..group_count {
+                let group_name = Bytes::from(read_redis_string(cursor)?);
+                let mut gld_ms = [0u8; 8];
+                let mut gld_seq = [0u8; 8];
+                cursor.read_exact(&mut gld_ms)?;
+                cursor.read_exact(&mut gld_seq)?;
+                let last_delivered_id = StreamId {
+                    ms: u64::from_le_bytes(gld_ms),
+                    seq: u64::from_le_bytes(gld_seq),
+                };
+                let (pel_count, _) = read_length(cursor)?;
+                check_alloc_bound(cursor, pel_count, 1, "stream_pel")?;
+                let mut pel = BTreeMap::new();
+                for _ in 0..pel_count {
+                    let mut pid_ms = [0u8; 8];
+                    let mut pid_seq = [0u8; 8];
+                    cursor.read_exact(&mut pid_ms)?;
+                    cursor.read_exact(&mut pid_seq)?;
+                    let pid = StreamId {
+                        ms: u64::from_le_bytes(pid_ms),
+                        seq: u64::from_le_bytes(pid_seq),
+                    };
+                    let consumer_name = Bytes::from(read_redis_string(cursor)?);
+                    let mut dt_buf = [0u8; 8];
+                    let mut dc_buf = [0u8; 8];
+                    cursor.read_exact(&mut dt_buf)?;
+                    cursor.read_exact(&mut dc_buf)?;
+                    pel.insert(
+                        pid,
+                        PendingEntry {
+                            consumer: consumer_name,
+                            delivery_time: u64::from_le_bytes(dt_buf),
+                            delivery_count: u64::from_le_bytes(dc_buf),
+                        },
+                    );
+                }
+                let (consumer_count, _) = read_length(cursor)?;
+                check_alloc_bound(cursor, consumer_count, 1, "stream_consumers")?;
+                let mut consumers = HashMap::new();
+                for _ in 0..consumer_count {
+                    let cname = Bytes::from(read_redis_string(cursor)?);
+                    let mut st_buf = [0u8; 8];
+                    cursor.read_exact(&mut st_buf)?;
+                    let seen_time = u64::from_le_bytes(st_buf);
+                    let (pending_count, _) = read_length(cursor)?;
+                    check_alloc_bound(cursor, pending_count, 1, "stream_pending")?;
+                    let mut pending = BTreeMap::new();
+                    for _ in 0..pending_count {
+                        let mut cid_ms = [0u8; 8];
+                        let mut cid_seq = [0u8; 8];
+                        cursor.read_exact(&mut cid_ms)?;
+                        cursor.read_exact(&mut cid_seq)?;
+                        pending.insert(
+                            StreamId {
+                                ms: u64::from_le_bytes(cid_ms),
+                                seq: u64::from_le_bytes(cid_seq),
+                            },
+                            (),
+                        );
+                    }
+                    consumers.insert(
+                        cname.clone(),
+                        Consumer {
+                            name: cname,
+                            pending,
+                            seen_time,
+                        },
+                    );
+                }
+                stream.groups.insert(
+                    group_name,
+                    ConsumerGroup {
+                        last_delivered_id,
+                        pel,
+                        consumers,
+                    },
+                );
+            }
+            let mut durable_buf = [0u8; 1];
+            cursor.read_exact(&mut durable_buf)?;
+            stream.durable = durable_buf[0] != 0;
+            let mut mdc_buf = [0u8; 4];
+            cursor.read_exact(&mut mdc_buf)?;
+            stream.max_delivery_count = u32::from_le_bytes(mdc_buf);
+
+            let mut entry = Entry::new_stream();
+            if let Some(rv) = entry.redis_value_mut() {
+                *rv = RedisValue::Stream(Box::new(stream));
+            }
+            entry
+        }
         other => bail!("Unsupported RDB type tag: {}", other),
     };
 
@@ -1447,6 +1640,7 @@ mod tests {
             (RDB_TYPE_SET, "set"),
             (RDB_TYPE_HASH, "hash"),
             (RDB_TYPE_ZSET_2, "zset"),
+            (RDB_TYPE_STREAM_MOON, "stream"),
         ] {
             let mut buf = Vec::new();
             write_length(&mut buf, u64::MAX); // lying count, zero element data follows
@@ -1458,4 +1652,135 @@ mod tests {
             );
         }
     }
+
+    /// Wave B stage 2b: `RDB_TYPE_STREAM_MOON` round-trips entries, PEL,
+    /// per-consumer pending sets, and the MQ durability flags — the fix for
+    /// the pre-existing `"__stream__:<len>"` placeholder that discarded all
+    /// of this on FULLRESYNC (a durable queue would come back as a plain
+    /// string, and MQ.POP/MQ.ACK would fail with a wrong-type error).
+    #[test]
+    fn test_stream_round_trip_full_fidelity() {
+        let mut stream = StreamData::new();
+        stream.durable = true;
+        stream.max_delivery_count = 5;
+        let id1 = StreamId { ms: 1000, seq: 0 };
+        let id2 = StreamId { ms: 1000, seq: 1 };
+        stream.add(
+            id1,
+            vec![(Bytes::from_static(b"f1"), Bytes::from_static(b"v1"))],
+        );
+        stream.add(
+            id2,
+            vec![(Bytes::from_static(b"f2"), Bytes::from_static(b"v2"))],
+        );
+        let mut pel = BTreeMap::new();
+        pel.insert(
+            id1,
+            PendingEntry {
+                consumer: Bytes::from_static(b"c1"),
+                delivery_time: 42,
+                delivery_count: 2,
+            },
+        );
+        let mut pending = BTreeMap::new();
+        pending.insert(id1, ());
+        let mut consumers = HashMap::new();
+        consumers.insert(
+            Bytes::from_static(b"c1"),
+            Consumer {
+                name: Bytes::from_static(b"c1"),
+                pending,
+                seen_time: 99,
+            },
+        );
+        stream.groups.insert(
+            Bytes::from_static(b"g1"),
+            ConsumerGroup {
+                last_delivered_id: id1,
+                pel,
+                consumers,
+            },
+        );
+
+        let mut entry = Entry::new_stream();
+        if let Some(rv) = entry.redis_value_mut() {
+            *rv = RedisValue::Stream(Box::new(stream));
+        }
+
+        let mut buf = Vec::new();
+        write_rdb_entry(&mut buf, b"mystream", &entry, 0);
+        assert_eq!(
+            buf[0], RDB_TYPE_STREAM_MOON,
+            "no TTL set, first byte is the type tag"
+        );
+
+        let mut cursor = Cursor::new(&buf[1..]);
+        let key = read_redis_string(&mut cursor).unwrap();
+        assert_eq!(key, b"mystream");
+        let loaded = read_rdb_entry(&mut cursor, RDB_TYPE_STREAM_MOON, None).unwrap();
+        match loaded.as_redis_value() {
+            RedisValueRef::Stream(s) => {
+                assert_eq!(s.entries.len(), 2);
+                assert_eq!(s.last_id, id2);
+                assert!(s.durable, "durable flag must survive round-trip");
+                assert_eq!(s.max_delivery_count, 5);
+                let g = s
+                    .groups
+                    .get(&Bytes::from_static(b"g1"))
+                    .expect("group g1 must survive round-trip");
+                assert_eq!(g.last_delivered_id, id1);
+                assert_eq!(g.pel.len(), 1);
+                assert_eq!(g.pel.get(&id1).unwrap().delivery_count, 2);
+                assert_eq!(g.consumers.len(), 1);
+                assert_eq!(
+                    g.consumers
+                        .get(&Bytes::from_static(b"c1"))
+                        .unwrap()
+                        .pending
+                        .len(),
+                    1
+                );
+            }
+            _ => panic!("expected Stream"),
+        }
+    }
+
+    /// Truncating a valid stream encoding at any prefix length must error
+    /// cleanly (never panic / OOB read) — same discipline as the CRC and
+    /// oversized-count regressions above.
+    #[test]
+    fn test_stream_truncated_input_rejected_not_panicking() {
+        let mut stream = StreamData::new();
+        stream.add(
+            StreamId { ms: 1, seq: 0 },
+            vec![(Bytes::from_static(b"f"), Bytes::from_static(b"v"))],
+        );
+        stream.groups.insert(
+            Bytes::from_static(b"g"),
+            ConsumerGroup {
+                last_delivered_id: StreamId { ms: 1, seq: 0 },
+                pel: BTreeMap::new(),
+                consumers: HashMap::new(),
+            },
+        );
+        let mut entry = Entry::new_stream();
+        if let Some(rv) = entry.redis_value_mut() {
+            *rv = RedisValue::Stream(Box::new(stream));
+        }
+        let mut full = Vec::new();
+        write_rdb_entry(&mut full, b"k", &entry, 0);
+
+        for cut in 1..full.len() {
+            let mut cursor = Cursor::new(&full[1..cut.max(1)]);
+            if cut <= 1 {
+                continue;
+            }
+            // Best-effort: skip the key first (may itself fail on tiny cuts).
+            if read_redis_string(&mut cursor).is_err() {
+                continue;
+            }
+            let _ = read_rdb_entry(&mut cursor, RDB_TYPE_STREAM_MOON, None);
+            // No panic reaching here is the assertion.
+        }
+    }
 }

From 268bf4876ab71092decb84c327bedc0eb6d1c687 Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:00:49 +0700
Subject: [PATCH 02/10] feat(mq): replication wire-framing constants +
 generalized apply engine

Preparatory plumbing for Wave B stage 2b MQ replication:

- src/mq/wal.rs: five synthetic MQ._REPL.CREATE/PUSH/POP/ACK/TRIGGER
  pseudo-command names plus is_mq_replay_command(), used to frame the
  existing encode_mq_*/decode_mq_* WAL payload bytes on the wire and to
  route them on the replica side (never dispatched to real clients).

- src/mq/trigger.rs: TriggerRegistry::iter(), needed to snapshot every
  registered trigger for the FULLRESYNC registry blob landing in a
  later commit.

- src/shard/shared_databases.rs: generalize apply_mq_create/push/pop/
  ack/trigger over a new MqApplyTarget trait (mq_databases_mut() /
  mq_durable_queue_registry_mut() / mq_trigger_registry_mut()),
  implemented for both ShardSliceInit (boot-time WAL replay) and
  ShardSlice (live replica apply, added next). One apply engine now
  serves both boot recovery and replica replication instead of two.

author: Tin Dang
---
 src/mq/trigger.rs             | 23 ++++++++++
 src/mq/wal.rs                 | 53 +++++++++++++++++++++
 src/shard/shared_databases.rs | 86 +++++++++++++++++++++++++++--------
 3 files changed, 144 insertions(+), 18 deletions(-)

diff --git a/src/mq/trigger.rs b/src/mq/trigger.rs
index e3d70b8c..474cb5a9 100644
--- a/src/mq/trigger.rs
+++ b/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"[..]]);
+    }
 }
diff --git a/src/mq/wal.rs b/src/mq/wal.rs
index d96efbe6..3c2f72cf 100644
--- a/src/mq/wal.rs
+++ b/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]
diff --git a/src/shard/shared_databases.rs b/src/shard/shared_databases.rs
index 03da395f..1cf0ce76 100644
--- a/src/shard/shared_databases.rs
+++ b/src/shard/shared_databases.rs
@@ -459,25 +459,75 @@ fn warn_skip_mq_record(shard_id: usize, kind: &str, payload: &[u8], stats: &mut
     }
 }
 
+/// Shared field accessors for the MQ apply engine, implemented by both the
+/// boot-time WAL-replay target (`ShardSliceInit`, single-threaded, pre-shard-
+/// spawn) and the live-replica-apply target (`ShardSlice`, the running
+/// shard's thread-local aggregate) — Wave B stage 2b. `durable_queue_registry`
+/// / `trigger_registry` / `databases` are identically named and typed on
+/// both structs (see `shard::slice`), so the trait is a thin adapter that
+/// lets `apply_mq_*` stay generic instead of existing twice: `replay_mq_wal`
+/// (boot) and `replication::apply::apply_mq` (live stream) share ONE engine.
+pub(crate) trait MqApplyTarget {
+    fn mq_databases_mut(&mut self) -> &mut [crate::storage::Database];
+    fn mq_durable_queue_registry_mut(
+        &mut self,
+    ) -> &mut Option<Box<crate::mq::DurableQueueRegistry>>;
+    fn mq_trigger_registry_mut(&mut self) -> &mut Option<Box<crate::mq::TriggerRegistry>>;
+}
+
+impl MqApplyTarget for crate::shard::slice::ShardSliceInit {
+    #[inline]
+    fn mq_databases_mut(&mut self) -> &mut [crate::storage::Database] {
+        &mut self.databases
+    }
+    #[inline]
+    fn mq_durable_queue_registry_mut(
+        &mut self,
+    ) -> &mut Option<Box<crate::mq::DurableQueueRegistry>> {
+        &mut self.durable_queue_registry
+    }
+    #[inline]
+    fn mq_trigger_registry_mut(&mut self) -> &mut Option<Box<crate::mq::TriggerRegistry>> {
+        &mut self.trigger_registry
+    }
+}
+
+impl MqApplyTarget for crate::shard::slice::ShardSlice {
+    #[inline]
+    fn mq_databases_mut(&mut self) -> &mut [crate::storage::Database] {
+        &mut self.databases
+    }
+    #[inline]
+    fn mq_durable_queue_registry_mut(
+        &mut self,
+    ) -> &mut Option<Box<crate::mq::DurableQueueRegistry>> {
+        &mut self.durable_queue_registry
+    }
+    #[inline]
+    fn mq_trigger_registry_mut(&mut self) -> &mut Option<Box<crate::mq::TriggerRegistry>> {
+        &mut self.trigger_registry
+    }
+}
+
 /// Apply a single decoded MqCreate record: registers the shard-level durable
 /// queue config AND (re)creates the stream in the record's OWN db (fixes the
 /// pre-K2 db-0 hardcode — MqCreate now carries `db_index`).
-fn apply_mq_create(
-    init: &mut crate::shard::slice::ShardSliceInit,
+pub(crate) fn apply_mq_create<T: MqApplyTarget>(
+    target: &mut T,
     db_index: usize,
     key: &[u8],
     max_delivery_count: u32,
 ) {
     let key_bytes = bytes::Bytes::copy_from_slice(key);
-    let reg = init
-        .durable_queue_registry
+    let reg = target
+        .mq_durable_queue_registry_mut()
         .get_or_insert_with(|| Box::new(crate::mq::DurableQueueRegistry::new()));
     reg.insert(
         key_bytes.clone(),
         crate::mq::DurableStreamConfig::new(key_bytes, max_delivery_count),
     );
 
-    if let Some(db) = init.databases.get_mut(db_index) {
+    if let Some(db) = target.mq_databases_mut().get_mut(db_index) {
         if let Ok(stream) = db.get_or_create_stream(key) {
             stream.durable = true;
             stream.max_delivery_count = max_delivery_count;
@@ -494,14 +544,14 @@ fn apply_mq_create(
 /// ORIGINAL assigned id. Idempotent — replaying the same id twice (e.g. a
 /// stream that already had this entry via a surviving RDB/rrdshard load) is
 /// a harmless no-op rather than double-counting `length`.
-fn apply_mq_push(
-    init: &mut crate::shard::slice::ShardSliceInit,
+pub(crate) fn apply_mq_push<T: MqApplyTarget>(
+    target: &mut T,
     db_index: usize,
     key: &[u8],
     id: crate::storage::stream::StreamId,
     fields: Vec<(bytes::Bytes, bytes::Bytes)>,
 ) {
-    if let Some(db) = init.databases.get_mut(db_index) {
+    if let Some(db) = target.mq_databases_mut().get_mut(db_index) {
         if let Ok(stream) = db.get_or_create_stream(key) {
             if !stream.entries.contains_key(&id) {
                 stream.add(id, fields);
@@ -517,8 +567,8 @@ fn apply_mq_push(
 /// for DLQ pushes is looked up from the main stream's `entries` map, which
 /// by this point already reflects every MqPush record replayed so far
 /// (records are applied strictly in WAL order).
-fn apply_mq_pop(
-    init: &mut crate::shard::slice::ShardSliceInit,
+pub(crate) fn apply_mq_pop<T: MqApplyTarget>(
+    target: &mut T,
     db_index: usize,
     key: &[u8],
     last_delivered: crate::storage::stream::StreamId,
@@ -530,7 +580,7 @@ fn apply_mq_pop(
 ) {
     use crate::storage::stream::{Consumer, PendingEntry};
 
-    let Some(db) = init.databases.get_mut(db_index) else {
+    let Some(db) = target.mq_databases_mut().get_mut(db_index) else {
         return;
     };
 
@@ -615,13 +665,13 @@ fn apply_mq_pop(
 /// for the old cursor-rollback-by-count heuristic. `Stream::xack` is
 /// idempotent (no-ops if the id isn't in the PEL), so replaying the same ack
 /// twice or acking an id that was never actually pending is harmless.
-fn apply_mq_ack(
-    init: &mut crate::shard::slice::ShardSliceInit,
+pub(crate) fn apply_mq_ack<T: MqApplyTarget>(
+    target: &mut T,
     db_index: usize,
     key: &[u8],
     id: crate::storage::stream::StreamId,
 ) {
-    if let Some(db) = init.databases.get_mut(db_index) {
+    if let Some(db) = target.mq_databases_mut().get_mut(db_index) {
         if let Ok(Some(stream)) = db.get_stream_mut(key) {
             let group_name = bytes::Bytes::from_static(MQ_GROUP_NAME);
             let _ = stream.xack(&group_name, &[id]);
@@ -633,8 +683,8 @@ fn apply_mq_ack(
 /// data. Registration is durable/replayed but NEVER fired during replay —
 /// firing only happens from live `MQ.PUSH` debounce arming
 /// (`src/shard/timers.rs::fire_pending_mq_triggers`).
-fn apply_mq_trigger(
-    init: &mut crate::shard::slice::ShardSliceInit,
+pub(crate) fn apply_mq_trigger<T: MqApplyTarget>(
+    target: &mut T,
     trig_key: Vec<u8>,
     queue_key: Vec<u8>,
     callback_cmd: Vec<u8>,
@@ -647,8 +697,8 @@ fn apply_mq_trigger(
         last_fire_ms: 0,
         pending_fire_ms: 0,
     };
-    let reg = init
-        .trigger_registry
+    let reg = target
+        .mq_trigger_registry_mut()
         .get_or_insert_with(|| Box::new(crate::mq::TriggerRegistry::new()));
     reg.register(bytes::Bytes::from(trig_key), entry);
 }

From 79b9c97f6fe8e298f82a77471cf603be77fcc4fc Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:01:11 +0700
Subject: [PATCH 03/10] feat(replication): MQ replica-side apply engine +
 FULLRESYNC snapshot leg
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Wave B stage 2b, replica-consuming half. Two pieces:

1. Live-record apply (src/replication/apply.rs): apply_local routes any
   MQ._REPL.* record to a new apply_mq(), which decodes the single
   bulk-string payload with the existing decode_mq_* functions and
   dispatches to the SAME apply_mq_* engine (shard::shared_databases,
   generalized last commit) boot-time WAL replay uses. A replica never
   fires MQ.TRIGGER callbacks on apply — registrations are stored as
   opaque data, matching boot-time replay.

2. FULLRESYNC snapshot leg (src/replication/mq_sync.rs, new): a
   per-shard MOON_AUX_MQ_REGISTRY blob carries the durable-queue
   registry and trigger registry — shard-level bookkeeping that lives
   outside the keyspace (Stream CONTENT rides the ordinary RDB body,
   covered by the redis_rdb.rs Stream codec fix). Wired at both master
   capture sites (single-shard PSYNC in master.rs, and the merged
   multi-shard path via a new PreparedShardSync.mq_blob field populated
   in spsc_handler.rs) and installed additively into every replica
   shard by load_snapshot (apply.rs), mirroring
   graph_sync::install_graph_store_many's precedent exactly: an EMPTY
   blob list (pre-MQ-replication master) warns-and-keeps local state; a
   non-empty list is authoritative and replaces it.

author: Tin Dang
---
 src/replication/apply.rs   | 152 ++++++++++++++
 src/replication/master.rs  |  24 +++
 src/replication/mod.rs     |   1 +
 src/replication/mq_sync.rs | 392 +++++++++++++++++++++++++++++++++++++
 src/shard/dispatch.rs      |   6 +
 src/shard/spsc_handler.rs  |   6 +
 6 files changed, 581 insertions(+)
 create mode 100644 src/replication/mq_sync.rs

diff --git a/src/replication/apply.rs b/src/replication/apply.rs
index c747c36a..3a8545a8 100644
--- a/src/replication/apply.rs
+++ b/src/replication/apply.rs
@@ -217,6 +217,16 @@ pub(crate) fn apply_local(
             return;
         }
 
+        // MQ.* effect records (Wave B stage 2b) arrive as one of the
+        // `MQ._REPL.*` synthetic pseudo-commands `shard::mq_exec` emits —
+        // never a real client command. Route through the SAME apply engine
+        // (`apply_mq_*` in `shard::shared_databases`) boot-time WAL replay
+        // uses, generalized over `MqApplyTarget` so both share one codec.
+        if crate::mq::wal::is_mq_replay_command(cmd) {
+            apply_mq(s, cmd, args);
+            return;
+        }
+
         // TEMPORAL.INVALIDATE arrives as the master's deterministic
         // wall-clock-pinned form (`TEMPORAL.INVALIDATE-AT <graph> <N|E>
         // <entity_id> <wall_ms>`, round-2 finding B) — apply with the SAME
@@ -422,6 +432,102 @@ fn apply_graph(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Fram
     }
 }
 
+/// Apply one replicated MQ effect record (Wave B stage 2b): decode the
+/// SINGLE bulk-string payload arg (the verbatim `encode_mq_*` bytes
+/// `shard::mq_exec::replicate_mq_record` shipped) and dispatch to the same
+/// `apply_mq_*` functions boot-time WAL replay uses
+/// (`shard::shared_databases`, generic over `MqApplyTarget` so this
+/// `ShardSlice` target and that `ShardSliceInit` target share one engine).
+///
+/// A replica NEVER fires `MQ.TRIGGER` callbacks on apply (registrations are
+/// stored as opaque data, same as boot-time replay — firing only happens
+/// 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,
+    };
+    use crate::shard::shared_databases::{
+        apply_mq_ack, apply_mq_create, apply_mq_pop, apply_mq_push, apply_mq_trigger,
+    };
+    use crate::storage::stream::StreamId;
+
+    let Some(Frame::BulkString(payload)) = args.first() else {
+        tracing::warn!(
+            "replica apply: malformed {} record — missing payload arg (diverges; \
+             full resync required)",
+            String::from_utf8_lossy(cmd)
+        );
+        return;
+    };
+
+    let applied = if cmd.eq_ignore_ascii_case(MQ_REPL_CREATE) {
+        decode_mq_create(payload).map(|(db_index, key, mdc)| {
+            apply_mq_create(s, db_index as usize, &key, mdc);
+        })
+    } else if cmd.eq_ignore_ascii_case(MQ_REPL_PUSH) {
+        decode_mq_push(payload).map(|(db_index, key, ms, seq, fields)| {
+            apply_mq_push(s, db_index as usize, &key, StreamId { ms, seq }, fields);
+        })
+    } else if cmd.eq_ignore_ascii_case(MQ_REPL_POP) {
+        decode_mq_pop(payload).map(|(db_index, key, last_delivered, claimed, dlq)| {
+            let claimed: Vec<(StreamId, u64)> = claimed
+                .into_iter()
+                .map(|(ms, seq, dc)| (StreamId { ms, seq }, dc))
+                .collect();
+            let dlq: Vec<(StreamId, StreamId)> = dlq
+                .into_iter()
+                .map(|(src_ms, src_seq, dlq_ms, dlq_seq)| {
+                    (
+                        StreamId {
+                            ms: src_ms,
+                            seq: src_seq,
+                        },
+                        StreamId {
+                            ms: dlq_ms,
+                            seq: dlq_seq,
+                        },
+                    )
+                })
+                .collect();
+            apply_mq_pop(
+                s,
+                db_index as usize,
+                &key,
+                StreamId {
+                    ms: last_delivered.0,
+                    seq: last_delivered.1,
+                },
+                claimed,
+                dlq,
+            );
+        })
+    } else if cmd.eq_ignore_ascii_case(MQ_REPL_ACK) {
+        decode_mq_ack(payload).map(|(db_index, key, ms, seq)| {
+            apply_mq_ack(s, db_index as usize, &key, StreamId { ms, seq });
+        })
+    } else if cmd.eq_ignore_ascii_case(MQ_REPL_TRIGGER) {
+        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 {
+        // Unreachable: `is_mq_replay_command` gated this call. Defensive.
+        tracing::debug!(
+            "replica apply: unrecognized MQ replay command {}",
+            String::from_utf8_lossy(cmd)
+        );
+        Some(())
+    };
+
+    if applied.is_none() {
+        tracing::warn!(
+            "replica apply: malformed {} record payload — skipped (MQ plane diverges; \
+             full resync required)",
+            String::from_utf8_lossy(cmd)
+        );
+    }
+}
+
 /// Apply a replicated `TEMPORAL.INVALIDATE-AT` record: same mutation the
 /// master ran (`apply_invalidate`) with the master's pinned `wall_ms`. The
 /// returned `GraphTemporal` WAL payload is dropped (`Ok(_)` is ignored),
@@ -577,6 +683,10 @@ pub(crate) fn load_snapshot(
     // Wave B ws-plane: exactly one blob regardless of the master's shard
     // count (shard-0-authoritative — see `ws_sync` module docs).
     let ws_registry_blob = redis_rdb::read_moon_aux(rdb, redis_rdb::MOON_AUX_WORKSPACE_REGISTRY);
+    // Wave B stage 2b: same per-shard-blob collection as graph — a
+    // multi-shard master's merged snapshot carries one MQ registry aux
+    // entry PER shard.
+    let mq_blobs = redis_rdb::read_moon_aux_all(rdb, redis_rdb::MOON_AUX_MQ_REGISTRY);
     let result: anyhow::Result<usize> = match crate::shard::slice::try_with_shard(|s| {
         for db in s.databases.iter_mut() {
             db.clear();
@@ -620,6 +730,48 @@ pub(crate) fn load_snapshot(
                 }
             }
         }
+        // Wave B stage 2b: install the master's MQ durable-queue + trigger
+        // registries (authoritative replace — an EMPTY blob list means the
+        // master sent no aux at all, a pre-MQ-replication master, so we
+        // warn-and-keep local state exactly like the graph fallback above;
+        // a non-empty list, even of all-zero-count blobs, is authoritative
+        // and clears local registries).
+        match &mq_blobs[..] {
+            blobs if !blobs.is_empty() => {
+                match crate::replication::mq_sync::install_mq_registry_many(s, blobs) {
+                    Some((queues, triggers)) => {
+                        if queues > 0 || triggers > 0 {
+                            tracing::info!(
+                                "replica snapshot: installed {} durable queue(s), {} \
+                                 trigger(s) from {} shard blob(s)",
+                                queues,
+                                triggers,
+                                blobs.len()
+                            );
+                        }
+                    }
+                    None => {
+                        return Err(anyhow::anyhow!(
+                            "replica snapshot: malformed MQ registry aux blob"
+                        ));
+                    }
+                }
+            }
+            _ => {
+                let has_local = s
+                    .durable_queue_registry
+                    .as_ref()
+                    .is_some_and(|r| !r.is_empty())
+                    || s.trigger_registry.as_ref().is_some_and(|r| !r.is_empty());
+                if has_local {
+                    tracing::warn!(
+                        "replica snapshot carried no MQ-registry aux but local durable \
+                         queue/trigger state exists — master predates MQ replication; \
+                         keeping local state (it may diverge)"
+                    );
+                }
+            }
+        }
         Ok(loaded)
     }) {
         Some(r) => r,
diff --git a/src/replication/master.rs b/src/replication/master.rs
index c85f8280..23688df6 100644
--- a/src/replication/master.rs
+++ b/src/replication/master.rs
@@ -748,6 +748,15 @@ pub async fn handle_psync_inline_single_shard(
                     let ws_registry_blob = crate::replication::ws_sync::export_workspace_registry(
                         shard_databases.workspace_registry().as_deref(),
                     );
+                    // Wave B stage 2b: this shard's MQ durable-queue +
+                    // trigger registry snapshot. ALWAYS written (empty blob
+                    // when both registries are unset) so the replica can
+                    // distinguish "master shard has no MQ state" from
+                    // "pre-MQ-replication master" (aux absent entirely).
+                    let mq_blob = crate::replication::mq_sync::export_mq_registry(
+                        s.durable_queue_registry.as_deref(),
+                        s.trigger_registry.as_deref(),
+                    );
                     let mut moon_aux: Vec<(&[u8], &[u8])> = Vec::new();
                     if let Some(ref v) = vec_defs {
                         moon_aux
@@ -765,6 +774,10 @@ pub async fn handle_psync_inline_single_shard(
                         crate::persistence::redis_rdb::MOON_AUX_WORKSPACE_REGISTRY,
                         &ws_registry_blob[..],
                     ));
+                    moon_aux.push((
+                        crate::persistence::redis_rdb::MOON_AUX_MQ_REGISTRY,
+                        &mq_blob[..],
+                    ));
                     crate::persistence::redis_rdb::write_rdb_refs_with_moon_aux(
                         &refs,
                         &moon_aux,
@@ -959,6 +972,7 @@ pub async fn handle_psync_inline_multi_shard(
     // `PreparedShardSync::ws_registry_blob`). "Keep the first Some" matches
     // the `vector_defs`/`text_defs` convention below.
     let mut ws_registry_blob: Option<Vec<u8>> = None;
+    let mut mq_blobs: Vec<Vec<u8>> = Vec::with_capacity(num_shards);
     for (shard, reply_rx) in reply_rxs {
         // Bounded wait (review): a wedged shard must not park this task —
         // and its registrations — forever. 30s is far past any observed
@@ -1002,6 +1016,7 @@ pub async fn handle_psync_inline_multi_shard(
         }
         #[cfg(feature = "graph")]
         graph_blobs.push(prepared.graph_blob);
+        mq_blobs.push(prepared.mq_blob);
         bodies.push(prepared.rdb_body);
     }
 
@@ -1027,6 +1042,15 @@ pub async fn handle_psync_inline_multi_shard(
             &w[..],
         ));
     }
+    // MQ registry state is per-shard (owner-hashed by queue/trigger key,
+    // same sharding model as graph names): one aux entry per shard, merged
+    // additively into every replica shard by `mq_sync::install_mq_registry_many`.
+    for blob in &mq_blobs {
+        moon_aux.push((
+            crate::persistence::redis_rdb::MOON_AUX_MQ_REGISTRY,
+            &blob[..],
+        ));
+    }
     let mut rdb_buf: Vec<u8> = Vec::new();
     crate::persistence::redis_rdb::write_rdb_merged(&moon_aux, &bodies, &mut rdb_buf);
     info!(
diff --git a/src/replication/mod.rs b/src/replication/mod.rs
index 8e906d50..a5fb9dc2 100644
--- a/src/replication/mod.rs
+++ b/src/replication/mod.rs
@@ -4,6 +4,7 @@ pub mod backlog;
 pub mod graph_sync;
 pub mod handshake;
 pub mod master;
+pub mod mq_sync;
 pub mod reason_del;
 pub mod replica;
 pub mod state;
diff --git a/src/replication/mq_sync.rs b/src/replication/mq_sync.rs
new file mode 100644
index 00000000..9dbe93eb
--- /dev/null
+++ b/src/replication/mq_sync.rs
@@ -0,0 +1,392 @@
+//! Wave B stage 2b: FULLRESYNC snapshot for a shard's MQ durable-queue +
+//! trigger registries — the `MOON_AUX_MQ_REGISTRY` aux blob, one per shard
+//! (mirrors [`crate::replication::graph_sync`]'s per-shard
+//! `MOON_AUX_GRAPH_STORE` blob).
+//!
+//! # What this blob does NOT carry
+//!
+//! Stream CONTENT (entries, `last_id`, consumer-group PEL, per-consumer
+//! pending sets) rides the ordinary keyspace RDB body — every durable
+//! queue's underlying `Stream` is an regular key, loaded by
+//! `redis_rdb::load_rdb` before this blob is installed (see
+//! `replication::apply::load_snapshot`). That codec (`RDB_TYPE_STREAM_MOON`)
+//! ALSO carries the `durable` flag and `max_delivery_count` on the stream
+//! value itself, so a freshly-loaded Stream is already fully functional for
+//! MQ.PUSH/POP/ACK without this blob.
+//!
+//! What lives OUTSIDE the keyspace — and is therefore missing without this
+//! blob — is the shard-level bookkeeping on `ShardSlice`:
+//! `durable_queue_registry` (used by MQ.CREATE idempotency / introspection)
+//! and `trigger_registry` (MQ.TRIGGER debounce state, never fired on a
+//! replica — see [`crate::replication::apply::apply_mq`]'s docs). Losing
+//! `trigger_registry` specifically means a replica-turned-master after
+//! failover would need every MQ.TRIGGER re-registered; this blob prevents
+//! that data loss.
+//!
+//! Blob format (version 1, little-endian):
+//! ```text
+//! [u8  version = 1]
+//! [u32 queue_count]
+//! per queue:
+//!   [u32 key_len][key bytes]
+//!   [u32 max_delivery_count]
+//! [u32 trigger_count]
+//! per trigger:
+//!   [u32 trig_key_len][trig_key bytes]
+//!   [u32 queue_key_len][queue_key bytes]
+//!   [u32 callback_len][callback bytes]
+//!   [u64 debounce_ms]
+//! ```
+//! `dlq_key` / `consumer_group` (on `DurableStreamConfig`) and
+//! `last_fire_ms` / `pending_fire_ms` (on `TriggerEntry`) are NOT
+//! serialized — both are deterministically derived / reset-on-install (see
+//! `DurableStreamConfig::new` and `apply_mq_trigger`'s doc comment), exactly
+//! like boot-time WAL replay already does.
+
+use bytes::Bytes;
+
+use crate::mq::{DurableQueueRegistry, DurableStreamConfig, TriggerEntry, TriggerRegistry};
+use crate::shard::slice::ShardSlice;
+
+const FORMAT_VERSION: u8 = 1;
+
+/// Export one shard's MQ registries as a snapshot blob. Always returns a
+/// blob — an empty shard (no durable queues, no triggers) encodes as
+/// `queue_count = 0, trigger_count = 0`, letting the replica distinguish
+/// "master shard has no MQ state" (authoritative: drop local registries)
+/// from "pre-MQ-replication master" (aux absent entirely, handled by the
+/// caller checking `blobs.is_empty()` exactly like graph does).
+pub fn export_mq_registry(
+    durable_queue_registry: Option<&DurableQueueRegistry>,
+    trigger_registry: Option<&TriggerRegistry>,
+) -> Vec<u8> {
+    let mut buf: Vec<u8> = Vec::with_capacity(64);
+    buf.push(FORMAT_VERSION);
+
+    let queue_count = durable_queue_registry.map_or(0, DurableQueueRegistry::len);
+    buf.extend_from_slice(&(queue_count as u32).to_le_bytes());
+    if let Some(reg) = durable_queue_registry {
+        for (key, config) in reg.iter() {
+            buf.extend_from_slice(&(key.len() as u32).to_le_bytes());
+            buf.extend_from_slice(key);
+            buf.extend_from_slice(&config.max_delivery_count.to_le_bytes());
+        }
+    }
+
+    let trigger_count = trigger_registry.map_or(0, TriggerRegistry::len);
+    buf.extend_from_slice(&(trigger_count as u32).to_le_bytes());
+    if let Some(reg) = trigger_registry {
+        for (trig_key, entry) in reg.iter() {
+            buf.extend_from_slice(&(trig_key.len() as u32).to_le_bytes());
+            buf.extend_from_slice(trig_key);
+            buf.extend_from_slice(&(entry.queue_key.len() as u32).to_le_bytes());
+            buf.extend_from_slice(&entry.queue_key);
+            buf.extend_from_slice(&(entry.callback_cmd.len() as u32).to_le_bytes());
+            buf.extend_from_slice(&entry.callback_cmd);
+            buf.extend_from_slice(&entry.debounce_ms.to_le_bytes());
+        }
+    }
+    buf
+}
+
+/// Install a MULTI-SHARD snapshot's MQ registry blobs — one per master
+/// shard, `read_moon_aux_all` order — into this replica shard's registries.
+/// Authoritative replace happens ONCE (drop all local queue/trigger state),
+/// then every blob installs additively. Queue keys and trigger keys are
+/// disjoint across blobs (each queue/trigger lives on exactly one master
+/// shard, hashed by key — same ownership contract graph's per-graph-name
+/// disjointness relies on); a duplicate key across blobs simply gets
+/// overwritten (last-wins), matching `DurableQueueRegistry::insert` /
+/// `TriggerRegistry::register`'s own semantics for a single blob.
+///
+/// Returns `Some((queues_installed, triggers_installed))`, or `None` on a
+/// malformed blob (registries are left in whatever partial state was
+/// reached — the caller aborts the sync and the replica retries with a
+/// fresh full resync, same contract as `graph_sync::install_graph_store_many`).
+pub fn install_mq_registry_many(s: &mut ShardSlice, blobs: &[Vec<u8>]) -> Option<(usize, usize)> {
+    s.durable_queue_registry = None;
+    s.trigger_registry = None;
+    let mut total_queues = 0usize;
+    let mut total_triggers = 0usize;
+    for blob in blobs {
+        let (q, t) = install_one(s, blob)?;
+        total_queues += q;
+        total_triggers += t;
+    }
+    Some((total_queues, total_triggers))
+}
+
+fn install_one(s: &mut ShardSlice, blob: &[u8]) -> Option<(usize, usize)> {
+    let mut cur = Cursor { data: blob, pos: 0 };
+    if cur.u8()? != FORMAT_VERSION {
+        return None;
+    }
+
+    let queue_count = cur.u32()? as usize;
+    for _ in 0..queue_count {
+        let key_len = cur.len_checked()? as usize;
+        let key = Bytes::copy_from_slice(cur.take(key_len)?);
+        let max_delivery_count = cur.u32()?;
+        let reg = s
+            .durable_queue_registry
+            .get_or_insert_with(|| Box::new(DurableQueueRegistry::new()));
+        reg.insert(
+            key.clone(),
+            DurableStreamConfig::new(key, max_delivery_count),
+        );
+    }
+
+    let trigger_count = cur.u32()? as usize;
+    for _ in 0..trigger_count {
+        let trig_key_len = cur.len_checked()? as usize;
+        let trig_key = Bytes::copy_from_slice(cur.take(trig_key_len)?);
+        let queue_key_len = cur.len_checked()? as usize;
+        let queue_key = Bytes::copy_from_slice(cur.take(queue_key_len)?);
+        let callback_len = cur.len_checked()? as usize;
+        let callback_cmd = Bytes::copy_from_slice(cur.take(callback_len)?);
+        let debounce_ms = cur.u64()?;
+        let reg = s
+            .trigger_registry
+            .get_or_insert_with(|| Box::new(TriggerRegistry::new()));
+        reg.register(
+            trig_key,
+            TriggerEntry {
+                queue_key,
+                callback_cmd,
+                debounce_ms,
+                last_fire_ms: 0,
+                pending_fire_ms: 0,
+            },
+        );
+    }
+
+    Some((queue_count, trigger_count))
+}
+
+/// Minimal bounds-checked reader over the blob (mirrors
+/// `graph_sync::Cursor` — kept as a private per-module copy rather than
+/// shared, matching that module's own precedent).
+struct Cursor<'a> {
+    data: &'a [u8],
+    pos: usize,
+}
+
+impl<'a> Cursor<'a> {
+    fn take(&mut self, n: usize) -> Option<&'a [u8]> {
+        let end = self.pos.checked_add(n)?;
+        if end > self.data.len() {
+            return None;
+        }
+        let s = &self.data[self.pos..end];
+        self.pos = end;
+        Some(s)
+    }
+    fn u8(&mut self) -> Option<u8> {
+        Some(self.take(1)?[0])
+    }
+    fn u32(&mut self) -> Option<u32> {
+        #[allow(clippy::unwrap_used)] // take(4) guarantees the length
+        Some(u32::from_le_bytes(self.take(4)?.try_into().unwrap()))
+    }
+    fn u64(&mut self) -> Option<u64> {
+        #[allow(clippy::unwrap_used)] // take(8) guarantees the length
+        Some(u64::from_le_bytes(self.take(8)?.try_into().unwrap()))
+    }
+    /// Read a `u32` length prefix, bounds-checked against the remaining
+    /// blob so a corrupt/adversarial length can't drive an oversized
+    /// `Bytes::copy_from_slice` allocation attempt before `take()` would
+    /// have rejected it anyway — same DoS-avoidance discipline as
+    /// `redis_rdb::check_alloc_bound`.
+    fn len_checked(&mut self) -> Option<u32> {
+        let len = self.u32()?;
+        if len as usize > self.data.len().saturating_sub(self.pos) {
+            return None;
+        }
+        Some(len)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::shard::shared_databases::ShardStoreMemory;
+    use crate::shard::slice::{ShardSlice, ShardSliceInit, init_shard};
+    use crate::storage::Database;
+    use crate::text::store::TextStore;
+    use crate::transaction::{DeferredHnswInserts, KvWriteIntents};
+    use crate::vector::store::VectorStore;
+    use std::sync::Arc;
+    use std::sync::atomic::AtomicUsize;
+
+    fn make_test_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(),
+            #[cfg(feature = "graph")]
+            graph_store: crate::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),
+            }),
+        })
+    }
+
+    #[test]
+    fn export_empty_registries_round_trips_to_zero_counts() {
+        let blob = export_mq_registry(None, None);
+        assert_eq!(blob[0], FORMAT_VERSION);
+        std::thread::spawn(|| {
+            init_shard(make_test_slice());
+            crate::shard::slice::with_shard(|s| {
+                let installed = install_mq_registry_many(s, &[blob]).expect("valid blob");
+                assert_eq!(installed, (0, 0));
+                assert!(
+                    s.durable_queue_registry.is_none() || {
+                        s.durable_queue_registry.as_ref().unwrap().is_empty()
+                    }
+                );
+            });
+        })
+        .join()
+        .unwrap();
+    }
+
+    #[test]
+    fn export_install_round_trip_preserves_queues_and_triggers() {
+        let mut queues = DurableQueueRegistry::new();
+        queues.insert(
+            Bytes::from_static(b"orders"),
+            DurableStreamConfig::new(Bytes::from_static(b"orders"), 5),
+        );
+        queues.insert(
+            Bytes::from_static(b"tasks"),
+            DurableStreamConfig::new(Bytes::from_static(b"tasks"), 3),
+        );
+        let mut triggers = TriggerRegistry::new();
+        triggers.register(
+            Bytes::from_static(b"wshex:orders"),
+            TriggerEntry {
+                queue_key: Bytes::from_static(b"orders"),
+                callback_cmd: Bytes::from_static(b"PUBLISH ch msg"),
+                debounce_ms: 1500,
+                last_fire_ms: 999, // transient — must NOT survive round-trip
+                pending_fire_ms: 888,
+            },
+        );
+
+        let blob = export_mq_registry(Some(&queues), Some(&triggers));
+
+        std::thread::spawn(move || {
+            init_shard(make_test_slice());
+            crate::shard::slice::with_shard(|s| {
+                // Pre-existing local state must be dropped (authoritative replace).
+                s.durable_queue_registry = Some(Box::new({
+                    let mut stale = DurableQueueRegistry::new();
+                    stale.insert(
+                        Bytes::from_static(b"stale"),
+                        DurableStreamConfig::new(Bytes::from_static(b"stale"), 1),
+                    );
+                    stale
+                }));
+
+                let (q, t) = install_mq_registry_many(s, &[blob]).expect("valid blob");
+                assert_eq!(q, 2);
+                assert_eq!(t, 1);
+
+                let qreg = s.durable_queue_registry.as_ref().unwrap();
+                assert!(qreg.get(b"stale").is_none(), "stale entry must be dropped");
+                assert_eq!(qreg.get(b"orders").unwrap().max_delivery_count, 5);
+                assert_eq!(qreg.get(b"tasks").unwrap().max_delivery_count, 3);
+
+                let treg = s.trigger_registry.as_ref().unwrap();
+                let entry = treg.get(b"wshex:orders").unwrap();
+                assert_eq!(entry.queue_key.as_ref(), b"orders");
+                assert_eq!(entry.callback_cmd.as_ref(), b"PUBLISH ch msg");
+                assert_eq!(entry.debounce_ms, 1500);
+                assert_eq!(entry.last_fire_ms, 0, "transient fire state must reset");
+                assert_eq!(entry.pending_fire_ms, 0, "transient fire state must reset");
+            });
+        })
+        .join()
+        .unwrap();
+    }
+
+    #[test]
+    fn install_many_merges_disjoint_shard_blobs() {
+        let mut q0 = DurableQueueRegistry::new();
+        q0.insert(
+            Bytes::from_static(b"q0"),
+            DurableStreamConfig::new(Bytes::from_static(b"q0"), 1),
+        );
+        let mut q1 = DurableQueueRegistry::new();
+        q1.insert(
+            Bytes::from_static(b"q1"),
+            DurableStreamConfig::new(Bytes::from_static(b"q1"), 2),
+        );
+        let blob0 = export_mq_registry(Some(&q0), None);
+        let blob1 = export_mq_registry(Some(&q1), None);
+
+        std::thread::spawn(move || {
+            init_shard(make_test_slice());
+            crate::shard::slice::with_shard(|s| {
+                let (q, _) = install_mq_registry_many(s, &[blob0, blob1]).expect("valid blobs");
+                assert_eq!(q, 2);
+                let reg = s.durable_queue_registry.as_ref().unwrap();
+                assert!(reg.get(b"q0").is_some());
+                assert!(reg.get(b"q1").is_some());
+            });
+        })
+        .join()
+        .unwrap();
+    }
+
+    #[test]
+    fn install_rejects_unknown_version() {
+        let mut blob = export_mq_registry(None, None);
+        blob[0] = 99;
+        std::thread::spawn(move || {
+            init_shard(make_test_slice());
+            crate::shard::slice::with_shard(|s| {
+                assert_eq!(install_mq_registry_many(s, &[blob]), None);
+            });
+        })
+        .join()
+        .unwrap();
+    }
+
+    #[test]
+    fn install_rejects_truncated_blob_not_panicking() {
+        let mut queues = DurableQueueRegistry::new();
+        queues.insert(
+            Bytes::from_static(b"orders"),
+            DurableStreamConfig::new(Bytes::from_static(b"orders"), 5),
+        );
+        let full = export_mq_registry(Some(&queues), None);
+        for cut in 0..full.len() {
+            let truncated = full[..cut].to_vec();
+            std::thread::spawn(move || {
+                init_shard(make_test_slice());
+                crate::shard::slice::with_shard(|s| {
+                    // Must not panic; either Some (if the cut happens to
+                    // still be a valid smaller blob) or None.
+                    let _ = install_mq_registry_many(s, &[truncated]);
+                });
+            })
+            .join()
+            .unwrap();
+        }
+    }
+}
diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs
index a427b5c6..5c5c0444 100644
--- a/src/shard/dispatch.rs
+++ b/src/shard/dispatch.rs
@@ -531,6 +531,12 @@ pub struct PreparedShardSync {
     /// other shard replies `None` and the merge loop keeps the first `Some`
     /// (same "keep shard 0's copy" convention as `vector_defs`/`text_defs`).
     pub ws_registry_blob: Option<Vec<u8>>,
+    /// This shard's MQ durable-queue + trigger registry snapshot blob (Wave
+    /// B stage 2b). MQ registry state is genuinely per-shard (owner-hashed
+    /// by queue/trigger key, same as graph names), so the stitcher writes
+    /// one `moon-mq-registry` aux entry PER shard and the replica imports
+    /// all of them additively — see `replication::mq_sync`.
+    pub mq_blob: Vec<u8>,
 }
 
 /// Messages sent to a shard via SPSC channels from the connection layer
diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs
index f9115d3a..3107620a 100644
--- a/src/shard/spsc_handler.rs
+++ b/src/shard/spsc_handler.rs
@@ -2866,6 +2866,7 @@ pub(crate) fn handle_shard_message_shared(
             let mut text_defs: Option<Vec<u8>> = None;
             #[cfg(feature = "graph")]
             let mut graph_blob: Vec<u8> = Vec::new();
+            let mut mq_blob: Vec<u8> = Vec::new();
             crate::shard::slice::with_shard(|s| {
                 let refs: Vec<&crate::storage::Database> = s.databases.iter().collect();
                 crate::persistence::redis_rdb::write_rdb_body_refs(&refs, &mut rdb_body);
@@ -2888,6 +2889,10 @@ pub(crate) fn handle_shard_message_shared(
                     graph_blob =
                         crate::replication::graph_sync::export_graph_store(&mut s.graph_store);
                 }
+                mq_blob = crate::replication::mq_sync::export_mq_registry(
+                    s.durable_queue_registry.as_deref(),
+                    s.trigger_registry.as_deref(),
+                );
             });
             // Wave B ws-plane: the workspace registry is process-global, not
             // per-shard, so only shard 0 (the sole writer of it — see
@@ -2934,6 +2939,7 @@ pub(crate) fn handle_shard_message_shared(
                 #[cfg(feature = "graph")]
                 graph_blob,
                 ws_registry_blob,
+                mq_blob,
             };
             if reply_tx.try_send(prepared).is_err() {
                 // The PSYNC task is gone (replica dropped mid-handshake) —

From e81f3bd8a19e2e6144ec0cbed8ff3f94c19c4ed2 Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:01:31 +0700
Subject: [PATCH 04/10] feat(replication): MQ master-side live-stream emission
 (Wave B stage 2b)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Emits MQ.* effect records to attached replicas at the moment they're
durably WAL-logged on the owner shard, closing the plane gap
warn_unreplicated_plane previously fail-loud-warned about for MQ.*
(the WS.* half of that warning is unaffected — a separate agent tracks
retiring it once WS replicates).

- src/replication/state.rs: record_local_write_global /
  record_local_write_db_global — ctx-free counterparts of
  record_local_write(_db) that resolve the same per-process
  ReplicationState Arc via admin::metrics_setup::get_global_repl_state_arc(),
  needed because shard::mq_exec runs on the shard's own OS thread
  without a ConnectionContext in scope. Also adds
  ReplicationState::num_shards().

- src/shard/mq_exec.rs: replicate_mq_record(), called from all five
  owner-shard handlers (CREATE/PUSH/POP/ACK/TRIGGER) right where each
  already builds its WAL payload — ships the IDENTICAL encoded bytes as
  one of the MQ._REPL.* pseudo-commands, gated single-shard-only
  (num_shards == 1), the same posture graph's own live path takes for
  its multi-shard gap. monoio-only; a no-op under runtime-tokio.

- src/server/conn/handler_monoio/txn.rs: MQ.PUBLISH's TXN-materialization
  leg replicates the same way at commit, on the connection's own thread
  (which — since owner == ctx.shard_id in this leg — IS the owning
  shard's thread).

- src/server/conn/handler_monoio/write.rs: retire the MQ half of
  warn_unreplicated_plane's call site now that MQ replicates in the
  single-shard configuration; the WS.CREATE/DROP call site is untouched.

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.

author: Tin Dang
---
 src/replication/state.rs                |  77 ++++++++++++++++
 src/server/conn/handler_monoio/ft.rs    |  26 ------
 src/server/conn/handler_monoio/txn.rs   |  26 +++++-
 src/server/conn/handler_monoio/write.rs |  15 ++--
 src/shard/mq_exec.rs                    | 114 ++++++++++++++++++++++--
 5 files changed, 218 insertions(+), 40 deletions(-)

diff --git a/src/replication/state.rs b/src/replication/state.rs
index b7e8bbd1..a2b447a5 100644
--- a/src/replication/state.rs
+++ b/src/replication/state.rs
@@ -235,6 +235,16 @@ impl ReplicationState {
             .unwrap_or(0)
     }
 
+    /// Number of shards this state tracks (`shard_offsets.len()`) — the
+    /// SAME value `ConnectionContext::num_shards` is initialized from, so a
+    /// ctx-free caller (e.g. `shard::mq_exec::replicate_mq_record`) can
+    /// evaluate the identical "single-shard only" gate the connection layer
+    /// uses (`ctx.num_shards == 1`) without a `ConnectionContext` in scope.
+    #[inline]
+    pub fn num_shards(&self) -> usize {
+        self.shard_offsets.len()
+    }
+
     /// Clone out a lock-free handle to the offset atomics.
     ///
     /// Called once per shard at event-loop startup; the handle is what the
@@ -351,6 +361,73 @@ pub fn fanout_hint_active() -> bool {
     FANOUT_HINT.load(Ordering::Relaxed)
 }
 
+/// Ctx-free twin of the connection layer's `record_local_write`
+/// (`server::conn::handler_monoio::ft::record_local_write`), for owner-shard
+/// execution paths that run on a shard's own OS thread WITHOUT a
+/// `ConnectionContext` in scope — today only `shard::mq_exec` (MQ.* effect-
+/// record replication, Wave B stage 2b).
+///
+/// Resolves the SAME per-process `ReplicationState` `Arc` the connection
+/// layer holds, via `admin::metrics_setup::get_global_repl_state_arc()` —
+/// both are clones of the one `Arc` built at boot (`main.rs` / `listener.rs`
+/// / `embedded.rs` all call `set_global_repl_state(repl_state.clone())` right
+/// after constructing it), so backlog append + offset advance land in the
+/// identical per-shard slot the connection-layer path would have used.
+/// No-op if replication was never initialized (embedded/library callers with
+/// no listener boot path).
+///
+/// Caller MUST be running on shard `shard_id`'s own OS thread — this pushes
+/// to `shard::self_msg`, which is a `thread_local!` queue (see that module's
+/// docs: "Tokio tasks must NOT push here").
+pub fn record_local_write_global(shard_id: usize, bytes: Bytes) {
+    let Some(repl_state_arc) = crate::admin::metrics_setup::get_global_repl_state_arc() else {
+        return;
+    };
+    let mut end_offset = u64::MAX;
+    if let Ok(g) = repl_state_arc.read() {
+        if let Some(slot) = g.per_shard_backlogs.get(shard_id) {
+            if let Some(backlog) = slot.lock().as_mut() {
+                backlog.append(&bytes);
+            }
+        }
+        end_offset = g.increment_shard_offset(shard_id, bytes.len() as u64);
+    }
+    crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout {
+        bytes,
+        end_offset,
+    });
+}
+
+/// Db-aware ctx-free twin of `record_local_write_db`, for the same
+/// ctx-free callers as [`record_local_write_global`]. Implements only the
+/// single-shard "prepend `SELECT` on db change" branch — ctx-free MQ
+/// replication is gated `num_shards == 1` by its caller (graph precedent:
+/// live streaming is single-shard-only), so the N-shard fused-SELECT branch
+/// (which needs `ctx.num_shards`) is never reached and intentionally not
+/// implemented here.
+pub fn record_local_write_db_global(shard_id: usize, db: usize, bytes: Bytes) {
+    let Some(repl_state_arc) = crate::admin::metrics_setup::get_global_repl_state_arc() else {
+        return;
+    };
+    let needs_select = repl_state_arc.read().is_ok_and(|g| {
+        g.stream_db.get(shard_id).is_some_and(|slot| {
+            if slot.load(Ordering::Relaxed) != db as i64 {
+                slot.store(db as i64, Ordering::Relaxed);
+                true
+            } else {
+                false
+            }
+        })
+    });
+    if needs_select {
+        record_local_write_global(
+            shard_id,
+            crate::persistence::aof::serialize_select_record(db),
+        );
+    }
+    record_local_write_global(shard_id, bytes);
+}
+
 /// Load replication IDs from {dir}/replication.state.
 /// If file missing or malformed, generates new IDs and saves them.
 pub fn load_replication_state(dir: &std::path::Path) -> (String, String) {
diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs
index 447d596c..ecd49b6a 100644
--- a/src/server/conn/handler_monoio/ft.rs
+++ b/src/server/conn/handler_monoio/ft.rs
@@ -164,32 +164,6 @@ fn serialize_select(db: usize) -> Vec<u8> {
     buf
 }
 
-/// Fail-loud marker for the MQ.* plane, which is NOT yet wired into
-/// replication (round-2 finding A): MQ.* writes persist durably on the master
-/// but never reach a replica — deterministic effect-record forms + replica
-/// apply arms are a follow-up (task #34, MQ half). Warn ONCE per process so
-/// operators running replicas learn about the divergence at write time
-/// instead of at failover. No-op (one Relaxed load) when no replica has ever
-/// attached.
-///
-/// Split from the former combined `warn_unreplicated_plane` (Wave B
-/// ws-plane): the WS half retired when WS.CREATE/DROP replication shipped —
-/// see `handler_monoio::write::try_handle_ws_command`, which no longer calls
-/// a warn function at all. This MQ half survives until MQ effect-record
-/// replication lands.
-pub(super) fn warn_mq_unreplicated(ctx: &ConnectionContext, cmd: &[u8]) {
-    use std::sync::atomic::{AtomicBool, Ordering};
-    static WARNED: AtomicBool = AtomicBool::new(false);
-    if replication_fanout_active(ctx) && !WARNED.swap(true, Ordering::Relaxed) {
-        tracing::warn!(
-            command = %String::from_utf8_lossy(cmd),
-            "replication: MQ.* writes are NOT replicated in v0.7 — a replica \
-             will not see this plane (known limitation, further occurrences \
-             not logged)"
-        );
-    }
-}
-
 /// Handle FT.* commands. Returns `true` if the command was consumed.
 ///
 /// Caller should `continue` the frame loop when this returns `true`.
diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs
index 620dd9cf..191cad89 100644
--- a/src/server/conn/handler_monoio/txn.rs
+++ b/src/server/conn/handler_monoio/txn.rs
@@ -167,10 +167,34 @@ pub(super) async fn try_handle_txn_commit(
                                     }
                                     payloads
                                 });
+                            // Wave B stage 2b: replicate the SAME payloads
+                            // (num_shards==1 gate — graph precedent) before
+                            // (or alongside) the WAL append, `ctx` is
+                            // available here (this leg runs on the
+                            // connection's own thread, which — since
+                            // `owner == ctx.shard_id` — IS the owning
+                            // shard's thread).
+                            let replicate =
+                                ctx.num_shards == 1 && super::ft::replication_fanout_active(ctx);
                             for payload in payloads {
+                                let payload_bytes = Bytes::from(payload);
+                                if replicate {
+                                    let frame = Frame::Array(
+                                        vec![
+                                            Frame::BulkString(Bytes::from_static(
+                                                crate::mq::wal::MQ_REPL_PUSH,
+                                            )),
+                                            Frame::BulkString(payload_bytes.clone()),
+                                        ]
+                                        .into(),
+                                    );
+                                    let record_bytes =
+                                        crate::persistence::aof::serialize_command(&frame);
+                                    super::ft::record_local_write_db(ctx, db_index, record_bytes);
+                                }
                                 crate::shard::mq_exec::wal_append_on_slice(
                                     crate::persistence::wal_v3::record::WalRecordType::MqPush,
-                                    Bytes::from(payload),
+                                    payload_bytes,
                                 );
                             }
                         } else {
diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs
index e5c3047a..1f1356ac 100644
--- a/src/server/conn/handler_monoio/write.rs
+++ b/src/server/conn/handler_monoio/write.rs
@@ -411,12 +411,15 @@ pub(super) async fn try_handle_mq_command(
         }
     };
 
-    // Round-2 finding A fail-loud: MQ mutations persist locally (WAL via
-    // execute_mq_on_owner) but are NOT replicated in v0.7 — surface the
-    // divergence once instead of letting a replica silently miss the plane.
-    if !sub.eq_ignore_ascii_case(b"LEN") && !sub.eq_ignore_ascii_case(b"DLQLEN") {
-        super::ft::warn_mq_unreplicated(ctx, cmd);
-    }
+    // Wave B stage 2b: MQ mutations now replicate live (see
+    // `shard::mq_exec::replicate_mq_record`) whenever `ctx.num_shards == 1`
+    // — the same single-shard gate graph's own live stream uses — so the
+    // round-2 `warn_mq_unreplicated` fail-loud no longer applies and is
+    // retired (its WS.* half was already retired by the WS-plane work).
+    // A multi-shard master still doesn't live-stream MQ writes (out of
+    // scope, tracked alongside graph's own un-warned multi-shard gap), but
+    // per graph precedent that gap is accepted silently rather than warned;
+    // a durable queue stays WAL-durable and FULLRESYNC-covered either way.
 
     if sub.eq_ignore_ascii_case(b"CREATE") {
         match validate_mq_create(cmd_args) {
diff --git a/src/shard/mq_exec.rs b/src/shard/mq_exec.rs
index 7e027b34..be663919 100644
--- a/src/shard/mq_exec.rs
+++ b/src/shard/mq_exec.rs
@@ -225,6 +225,78 @@ pub(crate) fn wal_append_on_slice(
     });
 }
 
+// ── Replication emission (Wave B stage 2b) ────────────────────────────────
+
+/// The current shard's index, read off the thread-local `ShardSlice`.
+/// `execute_mq_on_owner` always runs on a shard's own OS thread (either
+/// invoked locally by `write.rs::mq_hop_or_local` or via the `MqCommand`
+/// SPSC hop), so this is always available.
+#[inline]
+pub(crate) fn current_shard_id() -> usize {
+    crate::shard::slice::with_shard(|s| s.shard_id)
+}
+
+/// Replicate one MQ effect record to the live stream, mirroring the graph
+/// plane's single-shard gate
+/// (`server::conn::handler_monoio::write.rs`'s `record_local_write_db` call
+/// site: `num_shards == 1 && replication_fanout_active`). Ctx-free —
+/// `execute_mq_on_owner` runs on the shard's own OS thread without a
+/// `ConnectionContext` in scope, so this resolves the SAME per-process
+/// `ReplicationState` handle the connection layer uses via
+/// `replication::state::record_local_write_db_global` (see that function's
+/// docs for why sharing the global Arc is safe).
+///
+/// `cmd_name` is one of the `MQ._REPL.*` synthetic pseudo-commands
+/// (`crate::mq::wal::MQ_REPL_*`) — never dispatched to real clients, matched
+/// only by `replication::apply::apply_local`'s command-name routing (mirrors
+/// `GraphReplayCollector::is_graph_command`). `payload` is the SAME encoded
+/// WAL record bytes already built for `wal_append_on_slice`, so the
+/// replica's apply arm can call the identical `decode_mq_*` function WAL
+/// replay uses — one wire form, two consumers.
+///
+/// Multi-shard masters simply don't stream (the `num_shards == 1` gate),
+/// same durability-without-live-replication posture graph accepts for its
+/// own multi-shard gap — a durable queue is still WAL-durable and covered by
+/// FULLRESYNC, it just doesn't get incremental live updates yet.
+///
+/// monoio-only (`ShardSlice` / `shard::self_msg` are thread-per-core
+/// primitives — see `shard::self_msg` module docs); a no-op under
+/// `runtime-tokio`, where `execute_mq_on_owner` is unreachable in practice
+/// (`ShardSlice` is not the tokio runtime's storage model).
+#[cfg(feature = "runtime-monoio")]
+#[inline]
+fn replicate_mq_record(shard_id: usize, db_index: usize, cmd_name: &'static [u8], payload: &[u8]) {
+    if !crate::replication::state::fanout_hint_active() {
+        return;
+    }
+    let Some(repl_state) = crate::admin::metrics_setup::get_global_repl_state_arc() else {
+        return;
+    };
+    let single_shard = repl_state.read().is_ok_and(|g| g.num_shards() == 1);
+    if !single_shard {
+        return;
+    }
+    let frame = Frame::Array(
+        vec![
+            Frame::BulkString(Bytes::from_static(cmd_name)),
+            Frame::BulkString(Bytes::copy_from_slice(payload)),
+        ]
+        .into(),
+    );
+    let record_bytes = crate::persistence::aof::serialize_command(&frame);
+    crate::replication::state::record_local_write_db_global(shard_id, db_index, record_bytes);
+}
+
+#[cfg(not(feature = "runtime-monoio"))]
+#[inline]
+fn replicate_mq_record(
+    _shard_id: usize,
+    _db_index: usize,
+    _cmd_name: &'static [u8],
+    _payload: &[u8],
+) {
+}
+
 // ── Subcommand handlers ───────────────────────────────────────────────────────
 
 /// MQ.CREATE — owner creates the durable stream + registry entry.
@@ -270,6 +342,12 @@ fn handle_create(args: &[Frame], key_prefix: &Bytes, db_index: usize) -> Frame {
     {
         let payload =
             crate::mq::wal::encode_mq_create(db_index as u32, &eff_key, max_delivery_count);
+        replicate_mq_record(
+            current_shard_id(),
+            db_index,
+            crate::mq::wal::MQ_REPL_CREATE,
+            &payload,
+        );
         wal_append_on_slice(
             crate::persistence::wal_v3::record::WalRecordType::MqCreate,
             Bytes::from(payload),
@@ -322,8 +400,14 @@ fn handle_push(args: &[Frame], key_prefix: &Bytes, db_index: usize) -> Frame {
 
     match push_result {
         Ok(Some((msg_id, payload))) => {
-            // WAL: MqPush record on the owner shard (Wave B stage 2a) —
-            // durability plane only; no replication emission here (stage 2b).
+            // WAL: MqPush record on the owner shard (Wave B stage 2a), plus
+            // live replication fan-out of the SAME payload (Wave B stage 2b).
+            replicate_mq_record(
+                current_shard_id(),
+                db_index,
+                crate::mq::wal::MQ_REPL_PUSH,
+                &payload,
+            );
             wal_append_on_slice(
                 crate::persistence::wal_v3::record::WalRecordType::MqPush,
                 Bytes::from(payload),
@@ -506,6 +590,12 @@ fn handle_pop(args: &[Frame], key_prefix: &Bytes, db_index: usize) -> Frame {
         });
 
     if let Some(payload) = wal_payload {
+        replicate_mq_record(
+            current_shard_id(),
+            db_index,
+            crate::mq::wal::MQ_REPL_POP,
+            &payload,
+        );
         wal_append_on_slice(
             crate::persistence::wal_v3::record::WalRecordType::MqPop,
             Bytes::from(payload),
@@ -548,8 +638,10 @@ fn handle_ack(args: &[Frame], key_prefix: &Bytes, db_index: usize) -> Frame {
             // if a request id was never actually in the PEL (matches this
             // aggregate-count live path, which can't distinguish per-id
             // success without changing `Stream::xack`'s return type).
+            let shard_id = current_shard_id();
             for (ms, seq) in &msg_ids {
                 let payload = crate::mq::wal::encode_mq_ack(db_index as u32, &eff_key, *ms, *seq);
+                replicate_mq_record(shard_id, db_index, crate::mq::wal::MQ_REPL_ACK, &payload);
                 wal_append_on_slice(
                     crate::persistence::wal_v3::record::WalRecordType::MqAck,
                     Bytes::from(payload),
@@ -617,16 +709,24 @@ fn handle_trigger(args: &[Frame], key_prefix: &Bytes, db_index: usize) -> Frame
         reg.register(trig_key, entry);
     });
 
+    // `db_index` is unused by the TRIGGER *mutation* (registry is
+    // shard-level, not per-db) but is still passed through to
+    // `replicate_mq_record` for the SELECT-context invariant
+    // `record_local_write_db` documents: even a db-agnostic record must keep
+    // the replication stream's db context aligned with the writing
+    // connection's selected db, so the NEXT db-scoped record on this shard
+    // doesn't inherit a stale `SELECT`.
+    replicate_mq_record(
+        current_shard_id(),
+        db_index,
+        crate::mq::wal::MQ_REPL_TRIGGER,
+        &payload,
+    );
     wal_append_on_slice(
         crate::persistence::wal_v3::record::WalRecordType::MqTrigger,
         Bytes::from(payload),
     );
 
-    // `db_index` is unused for TRIGGER (registry is shard-level, not
-    // per-db) but kept in the signature for API uniformity. Suppress the
-    // unused-variable warning.
-    let _ = db_index;
-
     Frame::SimpleString(Bytes::from_static(b"OK"))
 }
 

From 17f521758d138aa307a4010c126bd1bd31ab0822 Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:02:42 +0700
Subject: [PATCH 05/10] test(mq): replication_mq.rs integration suite + 2 new
 fuzz targets
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

tests/replication_mq.rs, modeled on tests/replication_graph.rs +
tests/replication_planes.rs (single-shard master scope, matching
graph's own precedent):

- replica_syncs_mq_live_stream: MQ.CREATE/PUSH/ACK on the master reach
  an already-attached replica, verified via MQ.POP on the replica side
  (proves both registry state and message content replicated, not just
  the raw key) plus a stream-health assertion (zero reconnects).
- replica_backfills_mq_from_snapshot: a replica attaching to a master
  that already has MQ state backfills it from PSYNC FULLRESYNC (the
  redis_rdb.rs Stream codec fix is what makes this pass), then a live
  write after the snapshot boundary still lands.
- multi_shard_live_mq_write_not_streamed: pins the documented
  num_shards==1 live-stream gate down with an explicit RED-if-lifted
  test rather than leaving the limitation untested.

Two new cargo-fuzz targets (wired into both the PR and nightly
matrices in .github/workflows/fuzz.yml, per the project's "any new
parser/decoder needs a fuzz target" rule):

- mq_registry_blob: fuzzes install_mq_registry_many (the new
  MOON_AUX_MQ_REGISTRY blob decoder) on a throwaway per-iteration
  ShardSlice, mirroring mq_wal_record.rs's coverage of the sibling WAL
  op-blob decoders.
- redis_rdb_load: fuzzes persistence::redis_rdb::load_rdb directly
  (the FULLRESYNC RDB loader, including the new RDB_TYPE_STREAM_MOON
  tag) — distinct from the existing rdb_load.rs target, which covers
  the separate persistence::rdb SAVE/BGSAVE codec.

author: Tin Dang
---
 .github/workflows/fuzz.yml            |   4 +
 fuzz/Cargo.toml                       |  10 +
 fuzz/fuzz_targets/mq_registry_blob.rs |  69 +++++
 fuzz/fuzz_targets/redis_rdb_load.rs   |  22 ++
 tests/replication_mq.rs               | 371 ++++++++++++++++++++++++++
 5 files changed, 476 insertions(+)
 create mode 100644 fuzz/fuzz_targets/mq_registry_blob.rs
 create mode 100644 fuzz/fuzz_targets/redis_rdb_load.rs
 create mode 100644 tests/replication_mq.rs

diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml
index 305140d1..98fa9ff0 100644
--- a/.github/workflows/fuzz.yml
+++ b/.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
diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml
index 25f87862..649a595f 100644
--- a/fuzz/Cargo.toml
+++ b/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
diff --git a/fuzz/fuzz_targets/mq_registry_blob.rs b/fuzz/fuzz_targets/mq_registry_blob.rs
new file mode 100644
index 00000000..fb722bd8
--- /dev/null
+++ b/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),
+        }),
+    })
+}
diff --git a/fuzz/fuzz_targets/redis_rdb_load.rs b/fuzz/fuzz_targets/redis_rdb_load.rs
new file mode 100644
index 00000000..23f78152
--- /dev/null
+++ b/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);
+});
diff --git a/tests/replication_mq.rs b/tests/replication_mq.rs
new file mode 100644
index 00000000..51125a28
--- /dev/null
+++ b/tests/replication_mq.rs
@@ -0,0 +1,371 @@
+//! Wave B stage 2b: MQ-plane replication. A replica must synchronize durable
+//! queue / trigger-registry state from its master — both the live MQ.*
+//! mutation stream and the PSYNC full-resync snapshot backfill.
+//!
+//! Single-shard master scope (matches graph's own R0/R0.5 live-stream
+//! precedent — see `replication_graph.rs`). MQ effect records are streamed as
+//! synthetic `MQ._REPL.*` pseudo-commands carrying the SAME encoded payload
+//! bytes `shard::mq_exec::wal_append_on_slice` already builds for the WAL, so
+//! the replica's `apply_mq` reuses the identical `decode_mq_*` /
+//! `apply_mq_*` engine boot-time WAL replay uses (`shard::mq_exec::
+//! replicate_mq_record`, `replication::apply::apply_mq`). A multi-shard
+//! master does not live-stream MQ writes yet (the same `num_shards == 1` gate
+//! graph's own live path uses) — `multi_shard_live_mq_write_not_streamed`
+//! pins that limitation down explicitly rather than leaving it untested.
+//!
+//! Run: `MOON_BIN=./target/release/moon cargo test --test replication_mq -- --ignored`
+
+use std::io::{Read, Write};
+use std::net::TcpStream;
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+fn moon_bin() -> String {
+    std::env::var("MOON_BIN").unwrap_or_else(|_| "./target/release/moon".to_string())
+}
+
+fn start_moon(port: u16, dir: &str, shards: usize) -> Child {
+    start_moon_logged(port, dir, shards, None)
+}
+
+/// Like `start_moon`, optionally teeing stderr (tracing output) to a file so
+/// tests can assert stream-health properties — e.g. that the replica held ONE
+/// live connection instead of masking a dead stream behind a reconnect/
+/// full-resync loop (`replication_graph.rs` precedent).
+fn start_moon_logged(
+    port: u16,
+    dir: &str,
+    shards: usize,
+    stderr_log: Option<&std::path::Path>,
+) -> Child {
+    let stderr = match stderr_log {
+        Some(p) => Stdio::from(std::fs::File::create(p).expect("create log file")),
+        None => Stdio::null(),
+    };
+    Command::new(moon_bin())
+        .env("RUST_LOG", "moon=warn")
+        .args([
+            "--port",
+            &port.to_string(),
+            "--shards",
+            &shards.to_string(),
+            "--dir",
+            dir,
+            "--appendonly",
+            "no",
+            "--disk-free-min-pct",
+            "0",
+        ])
+        .stdout(Stdio::null())
+        .stderr(stderr)
+        .spawn()
+        .expect("Failed to start moon (set MOON_BIN to a built binary)")
+}
+
+/// Kill-on-drop guard so a panicking assertion never leaks a live server.
+struct Killer(Child);
+impl Drop for Killer {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+
+/// Send one inline command, return the raw reply text.
+fn send_cmd(addr: &str, cmd: &str) -> String {
+    let Ok(mut stream) = TcpStream::connect(addr) else {
+        return String::new();
+    };
+    stream.set_read_timeout(Some(Duration::from_secs(5))).ok();
+    if stream.write_all(format!("{cmd}\r\n").as_bytes()).is_err() {
+        return String::new();
+    }
+    read_quiet(&mut stream)
+}
+
+/// RESP array of bulk strings — binary-safe, for multiword MQ commands.
+fn send_resp(addr: &str, parts: &[&str]) -> String {
+    let Ok(mut stream) = TcpStream::connect(addr) else {
+        return String::new();
+    };
+    stream
+        .set_read_timeout(Some(Duration::from_millis(500)))
+        .ok();
+    let mut out = format!("*{}\r\n", parts.len()).into_bytes();
+    for p in parts {
+        out.extend_from_slice(format!("${}\r\n", p.len()).as_bytes());
+        out.extend_from_slice(p.as_bytes());
+        out.extend_from_slice(b"\r\n");
+    }
+    if stream.write_all(&out).is_err() {
+        return String::new();
+    }
+    read_quiet(&mut stream)
+}
+
+fn read_quiet(stream: &mut TcpStream) -> String {
+    let mut buf = Vec::new();
+    let mut chunk = [0u8; 4096];
+    let deadline = Instant::now() + Duration::from_millis(600);
+    while Instant::now() < deadline {
+        match stream.read(&mut chunk) {
+            Ok(0) => break,
+            Ok(n) => buf.extend_from_slice(&chunk[..n]),
+            Err(_) => {
+                if !buf.is_empty() {
+                    break;
+                }
+            }
+        }
+    }
+    String::from_utf8_lossy(&buf).into_owned()
+}
+
+/// Parse the payload out of a raw RESP bulk-string reply (`$<len>\r\n<data>\r\n`).
+/// `MQ PUSH` returns the assigned stream id as a `Frame::BulkString` (NOT an
+/// Integer/SimpleString), so callers that need the actual id value (e.g. to
+/// pass to a later `MQ ACK`) must parse it out rather than trim a `:` prefix.
+fn parse_bulk_string(raw: &str) -> Option<String> {
+    let mut lines = raw.split("\r\n");
+    let first = lines.next()?;
+    let len: i64 = first.strip_prefix('$')?.parse().ok()?;
+    if len < 0 {
+        return None; // $-1 nil
+    }
+    Some(lines.next()?.to_string())
+}
+
+fn wait_until<F: Fn() -> bool>(timeout: Duration, f: F) -> bool {
+    let deadline = Instant::now() + timeout;
+    while Instant::now() < deadline {
+        if f() {
+            return true;
+        }
+        std::thread::sleep(Duration::from_millis(150));
+    }
+    f()
+}
+
+/// REPL-MQ-01: live MQ.* mutations on the master reach the replica. The
+/// replica attaches to an EMPTY master, then MQ.CREATE / MQ.PUSH / MQ.ACK
+/// must all show up — isolating the live streaming leg from snapshot
+/// backfill.
+#[test]
+#[ignore]
+fn replica_syncs_mq_live_stream() {
+    let master_dir = tempfile::tempdir().unwrap();
+    let replica_dir = tempfile::tempdir().unwrap();
+    let master_addr = "127.0.0.1:16740";
+    let replica_addr = "127.0.0.1:16741";
+
+    let replica_log = replica_dir.path().join("replica-stderr.log");
+
+    let _master = Killer(start_moon(16740, master_dir.path().to_str().unwrap(), 1));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING")
+            .starts_with("+PONG")),
+        "master never became ready"
+    );
+    let _replica = Killer(start_moon_logged(
+        16741,
+        replica_dir.path().to_str().unwrap(),
+        1,
+        Some(&replica_log),
+    ));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING")
+            .starts_with("+PONG")),
+        "replica never became ready"
+    );
+    assert!(
+        send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16740").starts_with("+OK"),
+        "REPLICAOF failed"
+    );
+    // Let the handshake settle before the first write.
+    std::thread::sleep(Duration::from_millis(500));
+
+    // 1. MQ.CREATE + MQ.PUSH stream to the replica.
+    assert!(
+        send_cmd(master_addr, "MQ CREATE mq1 MAXDELIVERY 3").starts_with("+OK"),
+        "MQ CREATE failed"
+    );
+    let push = send_resp(master_addr, &["MQ", "PUSH", "mq1", "f1", "v1"]);
+    assert!(
+        parse_bulk_string(&push).is_some_and(|id| id.contains('-')),
+        "MQ PUSH must return a stream id: {push}"
+    );
+
+    // 2. The replica must be able to POP the pushed message — proving both
+    // the durable-queue registry (MQ.CREATE) and the message content
+    // (MQ.PUSH) replicated, not just the raw stream key.
+    let popped = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"]).contains("v1")
+    });
+    assert!(
+        popped,
+        "replica never saw the streamed MQ.PUSH message: {}",
+        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"])
+    );
+
+    // 3. A second push + ack on the master must also converge (ACK removes
+    // the entry from the PEL; DLQLEN stays 0 — proves MQ.ACK replicates,
+    // not just CREATE/PUSH).
+    let push2 = send_resp(master_addr, &["MQ", "PUSH", "mq1", "f2", "v2"]);
+    let id2 = parse_bulk_string(&push2).unwrap_or_default();
+    assert!(
+        id2.contains('-'),
+        "MQ PUSH #2 must return a stream id: {push2}"
+    );
+    let ack = send_resp(master_addr, &["MQ", "ACK", "mq1", &id2]);
+    assert!(ack.starts_with(':'), "MQ ACK failed: {ack}");
+
+    let acked_visible = wait_until(Duration::from_secs(10), || {
+        // After MQ.ACK replicates, a replica POP for the acked id's fields
+        // must NOT re-surface it (POP claims un-acked PEL entries fresh —
+        // the acked id was already removed from the group's PEL).
+        !send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "5"]).contains("v2")
+    });
+    assert!(
+        acked_visible,
+        "replica MQ.ACK never applied — acked message still POP-able"
+    );
+
+    // 4. STREAM-HEALTH: everything above must have arrived on ONE live
+    // stream (no reconnect/full-resync loop masking a dead live stream —
+    // `replication_graph.rs`'s REPL-GRAPH-01 precedent).
+    let log = std::fs::read_to_string(&replica_log).unwrap_or_default();
+    let reconnects = log.matches("reconnecting").count();
+    assert_eq!(
+        reconnects, 0,
+        "replica reconnected {reconnects}x — live stream did not hold:\n{log}"
+    );
+}
+
+/// REPL-MQ-02: a replica attaching to a master that ALREADY has MQ state
+/// (durable queue + pushed messages) must backfill it from the PSYNC
+/// full-resync snapshot — both the Stream CONTENT (RDB `RDB_TYPE_STREAM_MOON`
+/// fix landed alongside this stage) and the shard-level durable-queue
+/// registry (`MOON_AUX_MQ_REGISTRY` blob, `replication::mq_sync`).
+#[test]
+#[ignore]
+fn replica_backfills_mq_from_snapshot() {
+    let master_dir = tempfile::tempdir().unwrap();
+    let replica_dir = tempfile::tempdir().unwrap();
+    let master_addr = "127.0.0.1:16742";
+    let replica_addr = "127.0.0.1:16743";
+
+    let _master = Killer(start_moon(16742, master_dir.path().to_str().unwrap(), 1));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING")
+            .starts_with("+PONG")),
+        "master never became ready"
+    );
+
+    // Build MQ state on the master BEFORE any replica exists.
+    assert!(
+        send_cmd(master_addr, "MQ CREATE mqsnap MAXDELIVERY 5").starts_with("+OK"),
+        "seed MQ CREATE failed"
+    );
+    for (f, v) in [("f1", "v1"), ("f2", "v2"), ("f3", "v3")] {
+        let r = send_resp(master_addr, &["MQ", "PUSH", "mqsnap", f, v]);
+        assert!(
+            parse_bulk_string(&r).is_some_and(|id| id.contains('-')),
+            "seed MQ PUSH {f} failed: {r}"
+        );
+    }
+
+    // Now attach a fresh replica — it must learn mqsnap's durable-queue
+    // registration AND its 3 pushed messages purely from the snapshot.
+    let _replica = Killer(start_moon(16743, replica_dir.path().to_str().unwrap(), 1));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING")
+            .starts_with("+PONG")),
+        "replica never became ready"
+    );
+    assert!(
+        send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16742").starts_with("+OK"),
+        "REPLICAOF failed"
+    );
+
+    let backfilled = wait_until(Duration::from_secs(10), || {
+        let popped = send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "3"]);
+        popped.contains("v1") && popped.contains("v2") && popped.contains("v3")
+    });
+    assert!(
+        backfilled,
+        "replica never backfilled the 3 pre-existing snapshot messages: {}",
+        send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "3"])
+    );
+
+    // A live write AFTER the snapshot must also land (stream continues past
+    // the backfill boundary, same as REPL-GRAPH-02).
+    let r = send_resp(master_addr, &["MQ", "PUSH", "mqsnap", "f4", "v4"]);
+    assert!(
+        parse_bulk_string(&r).is_some_and(|id| id.contains('-')),
+        "post-snapshot MQ PUSH failed: {r}"
+    );
+    let grew = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "1"]).contains("v4")
+    });
+    assert!(grew, "post-snapshot live MQ.PUSH did not replicate");
+}
+
+/// REPL-MQ-03 (documented scope limitation): a multi-shard master does not
+/// live-stream MQ.* writes yet — `shard::mq_exec::replicate_mq_record` gates
+/// on `num_shards == 1`, the same posture graph's own live path takes for
+/// its multi-shard gap (durability is unaffected: the write is still
+/// WAL-durable and FULLRESYNC-covered, it just doesn't arrive incrementally).
+/// This test pins the gap down explicitly rather than leaving it silently
+/// untested, so a future multi-shard MQ live-stream fix has a RED test to
+/// flip to green instead of discovering the gap by accident.
+#[test]
+#[ignore]
+fn multi_shard_live_mq_write_not_streamed() {
+    let master_dir = tempfile::tempdir().unwrap();
+    let replica_dir = tempfile::tempdir().unwrap();
+    let master_addr = "127.0.0.1:16744";
+    let replica_addr = "127.0.0.1:16745";
+
+    let _master = Killer(start_moon(16744, master_dir.path().to_str().unwrap(), 2));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING")
+            .starts_with("+PONG")),
+        "master never became ready"
+    );
+    let _replica = Killer(start_moon(16745, replica_dir.path().to_str().unwrap(), 1));
+    assert!(
+        wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING")
+            .starts_with("+PONG")),
+        "replica never became ready"
+    );
+    assert!(
+        send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16744").starts_with("+OK"),
+        "REPLICAOF failed"
+    );
+    std::thread::sleep(Duration::from_millis(500));
+
+    assert!(
+        send_cmd(master_addr, "MQ CREATE mqms MAXDELIVERY 3").starts_with("+OK"),
+        "MQ CREATE failed"
+    );
+    let push = send_resp(master_addr, &["MQ", "PUSH", "mqms", "f1", "v1"]);
+    assert!(
+        parse_bulk_string(&push).is_some_and(|id| id.contains('-')),
+        "MQ PUSH failed: {push}"
+    );
+
+    // Short bounded wait (not the full 10s convergence window used by the
+    // positive tests) — a live-stream regression would show up within a
+    // couple of poll intervals; this is a negative assertion, not a
+    // performance timing test.
+    let never_arrived = !wait_until(Duration::from_secs(3), || {
+        send_resp(replica_addr, &["MQ", "POP", "mqms", "COUNT", "1"]).contains("v1")
+    });
+    assert!(
+        never_arrived,
+        "multi-shard MQ live write unexpectedly replicated — if this now \
+         passes, the num_shards==1 gate in `replicate_mq_record` was lifted; \
+         update this test's doc comment and promote it out of the \
+         known-limitation suite"
+    );
+}

From 06b3f7906aa728b0f2de0671249b7e7ae68a9fdc Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:02:47 +0700
Subject: [PATCH 06/10] docs(changelog): MQ-plane replication entry (Wave B
 stage 2b)

author: Tin Dang
---
 CHANGELOG.md | 45 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 45 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 77802002..ab66dd9a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -81,6 +81,51 @@ 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.* (the WS.* half of that warning is unaffected and still armed). 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.
+- 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)
 

From 9a48b327198bfaaf0334cd3ef30e6379e5fcde8a Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:30:56 +0700
Subject: [PATCH 07/10] docs(changelog): mixed-version FULLRESYNC operator note
 for Stream codec fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adversarial review of the mq-repl branch (verdict SHIP, P2 finding):
the RDB_TYPE_STREAM_MOON fix changes the PSYNC wire format for durable
Streams, so mixed-version pairs misbehave in both directions — old
replica vs new master hard-fails FULLRESYNC (reconnect-loops until
upgraded), new replica vs old master absorbs the legacy placeholder as
a corrupted string. Document the upgrade-together requirement where
operators will see it.

author: Tin Dang
---
 CHANGELOG.md | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab66dd9a..c491decc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -115,6 +115,13 @@ on the stage-2a MQ WAL effect records (PR #291):
   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

From cbbe38a9871b14d22bb2b000b9b9282ec161873a Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:42:57 +0700
Subject: [PATCH 08/10] docs: retire warn_mq_unreplicated remnants after WS+MQ
 plane composition
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Rebase-composition tidy: with both the WS plane (PR #293) and this
branch's MQ live stream shipped, the round-2 warn_unreplicated_plane
fail-loud marker has no surviving half — the renamed
warn_mq_unreplicated lost its last call site during the rebase, so the
function is removed (dead code under -D warnings) and the doc comment
in command/mq.rs plus the WS-plane CHANGELOG entry that still claimed
"warn_mq_unreplicated survives for MQ" are corrected.

author: Tin Dang
---
 CHANGELOG.md      | 11 +++++------
 src/command/mq.rs | 12 ++++++------
 2 files changed, 11 insertions(+), 12 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c491decc..19984408 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -72,21 +72,20 @@ 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.* (the WS.* half of that warning is unaffected and still armed). Builds
-on the stage-2a MQ WAL effect records (PR #291):
+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
diff --git a/src/command/mq.rs b/src/command/mq.rs
index 23659462..9a5f3105 100644
--- a/src/command/mq.rs
+++ b/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.

From 52bc9349efdf7481cb4a9b77dca70822948caa24 Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 12:55:01 +0700
Subject: [PATCH 09/10] test(mq): replica observation via reads +
 failover-promotion proof
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Rebase-composition fix: the replication_mq suite verified replica state
by issuing MQ POP against the replica, and readonly enforcement (PR
#292, merged after this branch was cut) now correctly rejects that with
-READONLY — MQ POP claims PEL entries and advances last_delivered, a
genuine write. The enforcement is right; the tests were exercising a
hole it closed.

Replica-side verification now uses read-only stream commands (XRANGE
for content, XPENDING summary for the consumer-group PEL), which also
makes the assertions STRONGER: the live-stream test now observes the
master's MqPop claim land in the replica PEL (1 pending) and the MqAck
drain it (0 pending) directly, instead of inferring both from what a
replica-local POP re-surfaced. Both tests then add the operational
failover proof — REPLICAOF NO ONE and a real MQ POP on the promoted
node — which exercises the replicated durable flag, consumer-group
state, and last_delivered cursor end-to-end the way an actual failover
would.

3/3 green locally (macOS release, --include-ignored).

author: Tin Dang
---
 tests/replication_mq.rs | 115 ++++++++++++++++++++++++++++++----------
 1 file changed, 87 insertions(+), 28 deletions(-)

diff --git a/tests/replication_mq.rs b/tests/replication_mq.rs
index 51125a28..01883094 100644
--- a/tests/replication_mq.rs
+++ b/tests/replication_mq.rs
@@ -195,42 +195,59 @@ fn replica_syncs_mq_live_stream() {
         "MQ PUSH must return a stream id: {push}"
     );
 
-    // 2. The replica must be able to POP the pushed message — proving both
-    // the durable-queue registry (MQ.CREATE) and the message content
-    // (MQ.PUSH) replicated, not just the raw stream key.
-    let popped = wait_until(Duration::from_secs(10), || {
-        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"]).contains("v1")
+    // 2. Read-only content check: `MQ POP` is a WRITE (claims PEL entries)
+    // and readonly enforcement (PR #292) correctly rejects it on a replica,
+    // so replica-side observation uses plain stream READS instead — XRANGE
+    // shows the pushed fields once MQ.CREATE + MQ.PUSH applied.
+    let seen = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["XRANGE", "mq1", "-", "+"]).contains("v1")
     });
     assert!(
-        popped,
+        seen,
         "replica never saw the streamed MQ.PUSH message: {}",
-        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"])
+        send_resp(replica_addr, &["XRANGE", "mq1", "-", "+"])
     );
 
-    // 3. A second push + ack on the master must also converge (ACK removes
-    // the entry from the PEL; DLQLEN stays 0 — proves MQ.ACK replicates,
-    // not just CREATE/PUSH).
-    let push2 = send_resp(master_addr, &["MQ", "PUSH", "mq1", "f2", "v2"]);
-    let id2 = parse_bulk_string(&push2).unwrap_or_default();
+    // 3. MqPop replication: the MASTER pops (claiming the entry into the
+    // consumer-group PEL and advancing last_delivered) — the replica's PEL,
+    // read via the XPENDING summary form, must converge to 1 pending.
+    let pop = send_resp(master_addr, &["MQ", "POP", "mq1", "COUNT", "1"]);
+    assert!(pop.contains("v1"), "master MQ POP failed: {pop}");
+    let pel_grew = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"]).starts_with("*4\r\n:1")
+    });
     assert!(
-        id2.contains('-'),
-        "MQ PUSH #2 must return a stream id: {push2}"
+        pel_grew,
+        "replica PEL never showed the master's claim: {}",
+        send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"])
     );
-    let ack = send_resp(master_addr, &["MQ", "ACK", "mq1", &id2]);
-    assert!(ack.starts_with(':'), "MQ ACK failed: {ack}");
 
-    let acked_visible = wait_until(Duration::from_secs(10), || {
-        // After MQ.ACK replicates, a replica POP for the acked id's fields
-        // must NOT re-surface it (POP claims un-acked PEL entries fresh —
-        // the acked id was already removed from the group's PEL).
-        !send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "5"]).contains("v2")
+    // 4. MqAck replication: acking the claimed id on the master must drain
+    // the replica's PEL back to 0.
+    let id1 = parse_bulk_string(&push).unwrap_or_default();
+    let ack = send_resp(master_addr, &["MQ", "ACK", "mq1", &id1]);
+    assert!(ack.starts_with(':'), "MQ ACK failed: {ack}");
+    let pel_drained = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"]).starts_with("*4\r\n:0")
     });
     assert!(
-        acked_visible,
-        "replica MQ.ACK never applied — acked message still POP-able"
+        pel_drained,
+        "replica MQ.ACK never applied — PEL still shows the acked entry: {}",
+        send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"])
     );
 
-    // 4. STREAM-HEALTH: everything above must have arrived on ONE live
+    // 5. A second push keeps streaming past the pop/ack cycle.
+    let push2 = send_resp(master_addr, &["MQ", "PUSH", "mq1", "f2", "v2"]);
+    assert!(
+        parse_bulk_string(&push2).is_some_and(|id| id.contains('-')),
+        "MQ PUSH #2 must return a stream id: {push2}"
+    );
+    let seen2 = wait_until(Duration::from_secs(10), || {
+        send_resp(replica_addr, &["XRANGE", "mq1", "-", "+"]).contains("v2")
+    });
+    assert!(seen2, "second MQ.PUSH never reached the replica");
+
+    // 6. STREAM-HEALTH: everything above must have arrived on ONE live
     // stream (no reconnect/full-resync loop masking a dead live stream —
     // `replication_graph.rs`'s REPL-GRAPH-01 precedent).
     let log = std::fs::read_to_string(&replica_log).unwrap_or_default();
@@ -239,6 +256,24 @@ fn replica_syncs_mq_live_stream() {
         reconnects, 0,
         "replica reconnected {reconnects}x — live stream did not hold:\n{log}"
     );
+
+    // 7. FAILOVER proof: promote the replica — a consumer must be able to
+    // POP the un-consumed message (v2) from it. This is the operational
+    // point of the whole plane: the replicated durable-flag, consumer-group
+    // state, and last_delivered cursor (advanced past v1 by the master's
+    // replicated pop) all have to be correct for the claim to work.
+    assert!(
+        send_cmd(replica_addr, "REPLICAOF NO ONE").starts_with("+OK"),
+        "promotion failed"
+    );
+    let promoted_pop = wait_until(Duration::from_secs(5), || {
+        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"]).contains("v2")
+    });
+    assert!(
+        promoted_pop,
+        "promoted replica could not POP the unconsumed message: {}",
+        send_resp(replica_addr, &["MQ", "POP", "mq1", "COUNT", "1"])
+    );
 }
 
 /// REPL-MQ-02: a replica attaching to a master that ALREADY has MQ state
@@ -287,14 +322,17 @@ fn replica_backfills_mq_from_snapshot() {
         "REPLICAOF failed"
     );
 
+    // Read-only observation (MQ POP is a write and is READONLY-rejected on
+    // a replica, per PR #292): the snapshot's stream content must appear
+    // via XRANGE.
     let backfilled = wait_until(Duration::from_secs(10), || {
-        let popped = send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "3"]);
-        popped.contains("v1") && popped.contains("v2") && popped.contains("v3")
+        let range = send_resp(replica_addr, &["XRANGE", "mqsnap", "-", "+"]);
+        range.contains("v1") && range.contains("v2") && range.contains("v3")
     });
     assert!(
         backfilled,
         "replica never backfilled the 3 pre-existing snapshot messages: {}",
-        send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "3"])
+        send_resp(replica_addr, &["XRANGE", "mqsnap", "-", "+"])
     );
 
     // A live write AFTER the snapshot must also land (stream continues past
@@ -305,9 +343,30 @@ fn replica_backfills_mq_from_snapshot() {
         "post-snapshot MQ PUSH failed: {r}"
     );
     let grew = wait_until(Duration::from_secs(10), || {
-        send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "1"]).contains("v4")
+        send_resp(replica_addr, &["XRANGE", "mqsnap", "-", "+"]).contains("v4")
     });
     assert!(grew, "post-snapshot live MQ.PUSH did not replicate");
+
+    // FAILOVER proof: a promoted replica must serve real consumers — POP
+    // has to claim all 4 messages, which exercises the snapshot-carried
+    // `durable` flag + consumer group (RDB_TYPE_STREAM_MOON codec) and the
+    // MQ registry aux blob end-to-end.
+    assert!(
+        send_cmd(replica_addr, "REPLICAOF NO ONE").starts_with("+OK"),
+        "promotion failed"
+    );
+    let promoted_pop = wait_until(Duration::from_secs(5), || {
+        let popped = send_resp(replica_addr, &["MQ", "POP", "mqsnap", "COUNT", "4"]);
+        popped.contains("v1")
+            && popped.contains("v2")
+            && popped.contains("v3")
+            && popped.contains("v4")
+    });
+    assert!(
+        promoted_pop,
+        "promoted replica could not POP the backfilled messages: {}",
+        send_resp(replica_addr, &["XRANGE", "mqsnap", "-", "+"])
+    );
 }
 
 /// REPL-MQ-03 (documented scope limitation): a multi-shard master does not

From dce1ed412601103e7bd0dc1bec2a572fa9d0bc0c Mon Sep 17 00:00:00 2001
From: Tin Dang <tindang.ht97@gmail.com>
Date: Sun, 12 Jul 2026 14:08:47 +0700
Subject: [PATCH 10/10] fix(replication): clamp MQ apply db index + assert
 truncation rejection
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two CodeRabbit findings on PR #294:

1. apply_mq used the payload's db index unclamped — a replica configured
   with fewer logical dbs than its master would silently no-op the
   stream mutation (MqApplyTarget::mq_databases_mut().get_mut(oob)
   returns None): CREATE registered the queue but never created the
   stream, PUSH/POP/ACK were dropped. Now clamped with the same posture
   (and debug log) as the generic KV clamp in apply_local. Boot-time WAL
   replay is unaffected — a shard replays its own records, whose db
   indices are valid by construction.

2. The RDB stream truncation test discarded the decode result, proving
   only "no panic". Every strict prefix of the length-prefixed encoding
   must hit EOF, so it now asserts is_err() at every cut point.

Gates: redis_rdb unit tests 30/30, replication_mq 3/3, clippy, fmt.

author: Tin Dang
---
 src/persistence/redis_rdb.rs |  9 ++++++--
 src/replication/apply.rs     | 40 ++++++++++++++++++++++++++++++++----
 2 files changed, 43 insertions(+), 6 deletions(-)

diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs
index 1e78ca5a..2673c6e2 100644
--- a/src/persistence/redis_rdb.rs
+++ b/src/persistence/redis_rdb.rs
@@ -1779,8 +1779,13 @@ mod tests {
             if read_redis_string(&mut cursor).is_err() {
                 continue;
             }
-            let _ = read_rdb_entry(&mut cursor, RDB_TYPE_STREAM_MOON, None);
-            // No panic reaching here is the assertion.
+            // Every strict prefix of the stream payload must be REJECTED —
+            // the encoding is a fixed sequence of length-prefixed fields, so
+            // a truncated read always hits EOF before completing.
+            assert!(
+                read_rdb_entry(&mut cursor, RDB_TYPE_STREAM_MOON, None).is_err(),
+                "truncated stream encoding at byte {cut} was accepted"
+            );
         }
     }
 }
diff --git a/src/replication/apply.rs b/src/replication/apply.rs
index 3a8545a8..409df96b 100644
--- a/src/replication/apply.rs
+++ b/src/replication/apply.rs
@@ -461,13 +461,40 @@ fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame])
         return;
     };
 
+    // Replicas configured with fewer logical dbs than the master clamp the
+    // payload's db index — same posture as the generic KV clamp in
+    // `apply_local` above. Without this, `MqApplyTarget::mq_databases_mut()
+    // .get_mut(out_of_range)` silently no-ops: CREATE would register the
+    // queue but never create the stream, and PUSH/POP/ACK would be dropped.
+    // (Boot-time WAL replay never needs this — a shard replays its OWN
+    // records, whose db indices are valid by construction.)
+    let db_count = s.databases.len();
+    fn clamp_mq_db(db_count: usize, db_index: u32) -> usize {
+        let idx = (db_index as usize).min(db_count.saturating_sub(1));
+        if idx != db_index as usize {
+            tracing::debug!(
+                "replica apply: MQ db {} clamped to {} ({} dbs on this shard)",
+                db_index,
+                idx,
+                db_count
+            );
+        }
+        idx
+    }
+
     let applied = if cmd.eq_ignore_ascii_case(MQ_REPL_CREATE) {
         decode_mq_create(payload).map(|(db_index, key, mdc)| {
-            apply_mq_create(s, db_index as usize, &key, mdc);
+            apply_mq_create(s, clamp_mq_db(db_count, db_index), &key, mdc);
         })
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_PUSH) {
         decode_mq_push(payload).map(|(db_index, key, ms, seq, fields)| {
-            apply_mq_push(s, db_index as usize, &key, StreamId { ms, seq }, fields);
+            apply_mq_push(
+                s,
+                clamp_mq_db(db_count, db_index),
+                &key,
+                StreamId { ms, seq },
+                fields,
+            );
         })
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_POP) {
         decode_mq_pop(payload).map(|(db_index, key, last_delivered, claimed, dlq)| {
@@ -492,7 +519,7 @@ fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame])
                 .collect();
             apply_mq_pop(
                 s,
-                db_index as usize,
+                clamp_mq_db(db_count, db_index),
                 &key,
                 StreamId {
                     ms: last_delivered.0,
@@ -504,7 +531,12 @@ fn apply_mq(s: &mut crate::shard::slice::ShardSlice, cmd: &[u8], args: &[Frame])
         })
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_ACK) {
         decode_mq_ack(payload).map(|(db_index, key, ms, seq)| {
-            apply_mq_ack(s, db_index as usize, &key, StreamId { ms, seq });
+            apply_mq_ack(
+                s,
+                clamp_mq_db(db_count, db_index),
+                &key,
+                StreamId { ms, seq },
+            );
         })
     } else if cmd.eq_ignore_ascii_case(MQ_REPL_TRIGGER) {
         decode_mq_trigger(payload).map(|(trig_key, queue_key, callback_cmd, debounce_ms)| {