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/CHANGELOG.md b/CHANGELOG.md index 77802002..19984408 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,16 +72,67 @@ on the multi-shard master path) is installed authoritatively at `load_snapshot`, same convention as the graph/vector/text aux blobs. New `ws_registry_record` cargo-fuzz target covers the blob decoder. -Split the former combined `warn_unreplicated_plane` fail-loud marker -(`handler_monoio/ft.rs`) into per-plane functions — the WS half is retired -now that this plane replicates; `warn_mq_unreplicated` survives for MQ, -which is not yet wired in. +The former combined `warn_unreplicated_plane` fail-loud marker +(`handler_monoio/ft.rs`) is fully retired: its WS half by this WS-plane +work, and its MQ half by the MQ-plane replication entry below. New `tests/replication_ws.rs`: live-stream parity (id + `created_at` round-trip through `WS.INFO`/`WS.AUTH` on the replica), snapshot-leg backfill, `WS.DROP` propagation, and a `--shards 4` leg exercising the shard-0 hop from connections that may land on any shard. +### Added — MQ-plane replication (Wave B stage 2b) + +Durable-queue MQ.* mutations now replicate to attached replicas, closing the +plane gap `warn_unreplicated_plane` previously fail-loud-warned about for +MQ.*. Builds on the stage-2a MQ WAL effect records (PR #291): + +- **Live stream**: `shard::mq_exec::replicate_mq_record` emits the SAME + encoded `MqCreate`/`MqPush`/`MqPop`/`MqAck`/`MqTrigger` payload bytes + already built for the WAL as one of five synthetic `MQ._REPL.*` + pseudo-commands (never dispatched to real clients), gated single-shard-only + (`num_shards == 1`) — the same posture graph's own live path takes for its + multi-shard gap. `MQ.PUBLISH`'s TXN-materialization leg + (`server::conn::handler_monoio::txn.rs`) replicates the same way at + commit. `replication::apply::apply_mq` decodes and applies through the + SAME `apply_mq_*` engine boot-time WAL replay uses + (`shard::shared_databases`, generalized over a new `MqApplyTarget` trait so + `ShardSliceInit` boot replay and `ShardSlice` live apply share one codec + instead of two). +- **FULLRESYNC**: a per-shard `MOON_AUX_MQ_REGISTRY` aux blob + (`replication::mq_sync`) carries the durable-queue registry and trigger + registry — the shard-level bookkeeping that lives outside the keyspace. + Installed additively into every replica shard + (`mq_sync::install_mq_registry_many`), mirroring + `graph_sync::install_graph_store_many`. +- **Fixed a pre-existing FULLRESYNC data-loss bug found along the way**: the + PSYNC RDB codec (`persistence::redis_rdb`, distinct from the + `persistence::rdb` SAVE/BGSAVE codec) serialized `Stream` values as a + placeholder `"__stream__:"` string, discarding all entries/PEL/ + consumer-group state. A full `RDB_TYPE_STREAM_MOON` (0xC8) codec — ported + from `persistence::rdb`'s battle-tested Stream serializer — now carries + entries, `last_id`, consumer groups (PEL + per-consumer pending), the + `durable` flag, and `max_delivery_count`. Without this fix, MQ FULLRESYNC + backfill would silently lose every durable queue's message content. + **Operator note — upgrade masters and replicas together** if any durable + Streams exist: a pre-fix replica syncing from a fixed master hard-fails + FULLRESYNC on the new `RDB_TYPE_STREAM_MOON` tag (and retries on its + reconnect loop until upgraded), while a fixed replica syncing from a + pre-fix master absorbs the old placeholder as a corrupted string value — + both inherent to fixing a wire-format bug, neither loses data already + durable in the master's WAL. +- Two new `cargo-fuzz` targets: `mq_registry_blob` (the new + `install_mq_registry_many` decoder) and `redis_rdb_load` (the FULLRESYNC + RDB loader as a whole, including the new Stream type tag) — wired into + both PR and nightly fuzz matrices. +- `tests/replication_mq.rs`: live-stream, snapshot-backfill, and a pinned + multi-shard-live-write-not-streamed limitation test (shards=1 scope, + matching graph's own precedent). + +A multi-shard master still does not live-stream MQ writes (durability is +unaffected — WAL + FULLRESYNC still cover it); tracked as a known follow-up +alongside graph's own multi-shard live-stream gap. + ### Fixed — docs site strict build red since 2026-07-08 (root-relative links) `docs/guides/tuning.md` (PR #245) and `docs/PRODUCTION-CONTRACT.md` (PR #263) 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 = + (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/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. 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 { + 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, Vec, Vec, 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 `. 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/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index 3b153bd7..2673c6e2 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__:"` 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, 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,140 @@ 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__:"` 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; + } + // 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 c747c36a..409df96b 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 // `, round-2 finding B) — apply with the SAME @@ -422,6 +432,134 @@ 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; + }; + + // 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, 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, + 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)| { + 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, + clamp_mq_db(db_count, db_index), + &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, + 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)| { + 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 +715,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 = match crate::shard::slice::try_with_shard(|s| { for db in s.databases.iter_mut() { db.clear(); @@ -620,6 +762,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> = None; + let mut mq_blobs: Vec> = 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 = 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 { + let mut buf: Vec = 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]) -> 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 { + Some(self.take(1)?[0]) + } + fn u32(&mut self) -> Option { + #[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 { + #[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 { + 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/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 { 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/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>, + /// 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, } /// Messages sent to a shard via SPSC channels from the connection layer 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")) } 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>; + fn mq_trigger_registry_mut(&mut self) -> &mut Option>; +} + +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> { + &mut self.durable_queue_registry + } + #[inline] + fn mq_trigger_registry_mut(&mut self) -> &mut Option> { + &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> { + &mut self.durable_queue_registry + } + #[inline] + fn mq_trigger_registry_mut(&mut self) -> &mut Option> { + &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( + 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( + 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( + 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( + 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( + target: &mut T, trig_key: Vec, queue_key: Vec, callback_cmd: Vec, @@ -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); } 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> = None; #[cfg(feature = "graph")] let mut graph_blob: Vec = Vec::new(); + let mut mq_blob: Vec = 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) — diff --git a/tests/replication_mq.rs b/tests/replication_mq.rs new file mode 100644 index 00000000..01883094 --- /dev/null +++ b/tests/replication_mq.rs @@ -0,0 +1,430 @@ +//! 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 (`$\r\n\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 { + 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 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. 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!( + seen, + "replica never saw the streamed MQ.PUSH message: {}", + send_resp(replica_addr, &["XRANGE", "mq1", "-", "+"]) + ); + + // 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!( + pel_grew, + "replica PEL never showed the master's claim: {}", + send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"]) + ); + + // 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!( + pel_drained, + "replica MQ.ACK never applied — PEL still shows the acked entry: {}", + send_resp(replica_addr, &["XPENDING", "mq1", "__mq_consumers"]) + ); + + // 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(); + let reconnects = log.matches("reconnecting").count(); + assert_eq!( + 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 +/// (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" + ); + + // 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 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, &["XRANGE", "mqsnap", "-", "+"]) + ); + + // 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, &["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 +/// 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" + ); +}