From c1c3146829c699a6bbc4836145c1724f991f0826 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:50:34 +0700 Subject: [PATCH 1/9] feat(workspace): RESP wire format for WS.CREATE/DROP replication records MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane groundwork. Adds `workspace::repl` — thin RESP-array wrapping of the existing WAL payload bytes (`encode_workspace_create`/`encode_workspace_drop`) into two internal pseudo-commands, `WS.CREATE.APPLY` / `WS.DROP.APPLY`, so the exact bytes already written to shard 0's WAL can flow over the replication wire and be recognized by the replica apply path before generic dispatch (mirrors `GRAPH.ADDNODE` and friends). No new encoding is invented: the payload is always the verbatim output of the WAL encoders. Not yet wired into any write path or replica apply arm — that lands in a follow-up commit alongside the shard-0 affinity hop. author: Tin Dang --- src/workspace/mod.rs | 1 + src/workspace/repl.rs | 146 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+) create mode 100644 src/workspace/repl.rs diff --git a/src/workspace/mod.rs b/src/workspace/mod.rs index 4ebe3e01..ceae3e75 100644 --- a/src/workspace/mod.rs +++ b/src/workspace/mod.rs @@ -6,6 +6,7 @@ //! via `extract_hash_tag()` in `shard::dispatch`. pub mod registry; +pub mod repl; pub mod wal; use bytes::Bytes; diff --git a/src/workspace/repl.rs b/src/workspace/repl.rs new file mode 100644 index 00000000..917b4408 --- /dev/null +++ b/src/workspace/repl.rs @@ -0,0 +1,146 @@ +//! Replication wire format for WS.CREATE / WS.DROP (Wave B ws-plane). +//! +//! The workspace registry is process-global (not per-key), so its mutations +//! cannot ride the generic KV replication stream (which replays user-syntax +//! commands through `dispatch()`). Two internal-only pseudo-commands carry +//! the WAL payload bytes verbatim over the SAME RESP wire the KV/graph +//! streams use (`record_local_write` / `drain_replicated_commands`): +//! +//! ```text +//! WS.CREATE.APPLY payload = wal::encode_workspace_create(..) +//! WS.DROP.APPLY payload = wal::encode_workspace_drop(..) +//! ``` +//! +//! Reusing the exact WAL byte layout (rather than inventing a second encoding) +//! means the replica applies the SAME `ws_id` + `name` + `created_at_ms` the +//! master's WAL persisted — no separate serializer to keep in sync, no risk of +//! the wire form drifting from the on-disk form. `apply::apply_local` +//! recognizes both names before generic dispatch (mirroring +//! `GraphReplayCollector::is_graph_command`); neither name is a real client +//! command (`is_ws_command` / the `phf` dispatch table never see them). +//! +//! Never emitted with a re-minted UUID: `WS.CREATE.APPLY`'s payload always +//! carries the MASTER's `ws_id` (UUIDv7 is nondeterministic — a replica must +//! never allocate its own). + +use bytes::Bytes; + +/// Internal-only command name for a replicated WorkspaceCreate record. Never +/// dispatched from a client connection. +pub const WS_CREATE_APPLY_CMD: &[u8] = b"WS.CREATE.APPLY"; +/// Internal-only command name for a replicated WorkspaceDrop record. +pub const WS_DROP_APPLY_CMD: &[u8] = b"WS.DROP.APPLY"; + +/// Write a RESP bulk string element: `$\r\n\r\n` +fn write_bulk(buf: &mut Vec, data: &[u8]) { + buf.push(b'$'); + buf.extend_from_slice(itoa::Buffer::new().format(data.len()).as_bytes()); + buf.extend_from_slice(b"\r\n"); + buf.extend_from_slice(data); + buf.extend_from_slice(b"\r\n"); +} + +/// Write a RESP array header: `*\r\n` +fn write_array_header(buf: &mut Vec, count: usize) { + buf.push(b'*'); + buf.extend_from_slice(itoa::Buffer::new().format(count).as_bytes()); + buf.extend_from_slice(b"\r\n"); +} + +/// Serialize `WS.CREATE.APPLY ` as a RESP array. `payload` is the +/// exact byte output of [`super::wal::encode_workspace_create`]. +pub fn serialize_ws_create_apply(payload: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(32 + payload.len()); + write_array_header(&mut buf, 2); + write_bulk(&mut buf, WS_CREATE_APPLY_CMD); + write_bulk(&mut buf, payload); + buf +} + +/// Serialize `WS.DROP.APPLY ` as a RESP array. `payload` is the +/// exact byte output of [`super::wal::encode_workspace_drop`]. +pub fn serialize_ws_drop_apply(payload: &[u8]) -> Vec { + let mut buf = Vec::with_capacity(32 + payload.len()); + write_array_header(&mut buf, 2); + write_bulk(&mut buf, WS_DROP_APPLY_CMD); + write_bulk(&mut buf, payload); + buf +} + +/// Extract the single bulk-string payload argument from a parsed +/// `WS.CREATE.APPLY` / `WS.DROP.APPLY` frame's args (i.e. everything after the +/// command name). Returns `None` if the arg is missing or not a bulk/simple +/// string — a malformed or truncated replication record. +pub fn extract_payload(args: &[crate::protocol::Frame]) -> Option { + use crate::protocol::Frame; + match args.first() { + Some(Frame::BulkString(b) | Frame::SimpleString(b)) => Some(b.clone()), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::{ParseConfig, parse}; + use bytes::BytesMut; + + /// Parse RESP bytes back into a `Frame` the same way the replica's + /// `drain_replicated_commands` does, and return (cmd, args). + fn parse_command(bytes: &[u8]) -> (Vec, Vec) { + use crate::protocol::Frame; + let mut buf = BytesMut::from(bytes); + let config = ParseConfig::default(); + let frame = parse::parse(&mut buf, &config) + .expect("parse ok") + .expect("complete frame"); + match frame { + Frame::Array(arr) => { + let mut it = arr.into_iter(); + let cmd = match it.next() { + Some(Frame::BulkString(b) | Frame::SimpleString(b)) => b.to_vec(), + _ => panic!("expected command name"), + }; + (cmd, it.collect()) + } + other => panic!("expected array frame, got {other:?}"), + } + } + + #[test] + fn ws_create_apply_round_trips_through_resp_parser() { + let ws_id = [7u8; 16]; + let payload = + crate::workspace::wal::encode_workspace_create(&ws_id, b"acme", 1_752_000_000_000); + let wire = serialize_ws_create_apply(&payload); + + let (cmd, args) = parse_command(&wire); + assert_eq!(cmd, WS_CREATE_APPLY_CMD); + let decoded_payload = extract_payload(&args).expect("payload arg present"); + let (decoded_id, decoded_name, decoded_created_at) = + crate::workspace::wal::decode_workspace_create(&decoded_payload) + .expect("valid payload"); + assert_eq!(decoded_id, ws_id); + assert_eq!(decoded_name, b"acme"); + assert_eq!(decoded_created_at, 1_752_000_000_000); + } + + #[test] + fn ws_drop_apply_round_trips_through_resp_parser() { + let ws_id = [9u8; 16]; + let payload = crate::workspace::wal::encode_workspace_drop(&ws_id); + let wire = serialize_ws_drop_apply(&payload); + + let (cmd, args) = parse_command(&wire); + assert_eq!(cmd, WS_DROP_APPLY_CMD); + let decoded_payload = extract_payload(&args).expect("payload arg present"); + let decoded_id = + crate::workspace::wal::decode_workspace_drop(&decoded_payload).expect("valid payload"); + assert_eq!(decoded_id, ws_id); + } + + #[test] + fn extract_payload_missing_arg_returns_none() { + assert!(extract_payload(&[]).is_none()); + } +} From 46548fe870cfc3a8a9bd00c8bd10487ee9aa6922 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:51:22 +0700 Subject: [PATCH 2/9] feat(replication): versioned WS registry PSYNC snapshot blob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane groundwork. Adds `replication::ws_sync` — export/install of the process-global `WorkspaceRegistry` as a new `MOON_AUX_WORKSPACE_REGISTRY` RDB aux blob (`persistence::redis_rdb`), for the FULLRESYNC full-resync leg. Unlike the per-shard `MOON_AUX_GRAPH_STORE`/vector/text blobs, the registry is a single process-global structure (one `Mutex` on `ShardDatabases`, visible from every shard's thread) rather than sharded state, so this blob is singular and shard-0-authoritative: exactly one blob belongs in the merged RDB regardless of `--shards`. Versioned little-endian format (version byte + entry count + per-entry ws_id/name/created_at), bounds- checked decoder returning `None` on any malformed input (truncated header, truncated entry, unknown version) rather than panicking — a corrupted or adversarial PSYNC stream must never crash a replica. `None` and `Some(empty)` registries encode identically so a replica can distinguish "master authoritatively has zero workspaces" from "pre-WS-sync master, aux absent entirely" once wired into `load_snapshot`. Includes 5 unit tests (round-trip, none/empty equivalence, 4 malformed-input shapes, truncated-full-blob, empty-name/negative-created_at edge case) and a new `ws_registry_record` cargo-fuzz target for the decoder (Testing rule: every new decoder needs a fuzz target). Not yet wired into any PSYNC capture site or `load_snapshot` — that lands in a follow-up commit. author: Tin Dang --- fuzz/Cargo.toml | 5 + fuzz/fuzz_targets/ws_registry_record.rs | 18 +++ src/persistence/redis_rdb.rs | 6 + src/replication/mod.rs | 1 + src/replication/ws_sync.rs | 195 ++++++++++++++++++++++++ 5 files changed, 225 insertions(+) create mode 100644 fuzz/fuzz_targets/ws_registry_record.rs create mode 100644 src/replication/ws_sync.rs diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index 41501e78..25f87862 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -88,3 +88,8 @@ doc = false name = "mq_wal_record" path = "fuzz_targets/mq_wal_record.rs" doc = false + +[[bin]] +name = "ws_registry_record" +path = "fuzz_targets/ws_registry_record.rs" +doc = false diff --git a/fuzz/fuzz_targets/ws_registry_record.rs b/fuzz/fuzz_targets/ws_registry_record.rs new file mode 100644 index 00000000..81636dee --- /dev/null +++ b/fuzz/fuzz_targets/ws_registry_record.rs @@ -0,0 +1,18 @@ +#![no_main] +use libfuzzer_sys::fuzz_target; + +use moon::replication::ws_sync::install_workspace_registry; + +/// Fuzz the WS-plane PSYNC snapshot decoder (`ws_sync::install_workspace_registry`, +/// Wave B ws-plane). +/// +/// `data` is untrusted bytes taken straight from a `MOON_AUX_WORKSPACE_REGISTRY` +/// RDB aux blob sent by a peer over PSYNC full-resync — a malicious or corrupted +/// "master" must never crash a replica applying it. `install_workspace_registry` +/// is documented to return `None` on any malformed input (bad version, truncated +/// count, truncated entry, truncated name) rather than panic or read out of +/// bounds; any panic/OOB here is a bug. Mirrors `graph_props_record.rs` and +/// `wal_v3_record.rs`. +fuzz_target!(|data: &[u8]| { + let _ = install_workspace_registry(data); +}); diff --git a/src/persistence/redis_rdb.rs b/src/persistence/redis_rdb.rs index 72bb5630..3b153bd7 100644 --- a/src/persistence/redis_rdb.rs +++ b/src/persistence/redis_rdb.rs @@ -449,6 +449,12 @@ pub const MOON_AUX_TEXT_DEFS: &[u8] = b"moon-text-defs"; /// (`replication::graph_sync::export_graph_store` blob — frozen CSR segment /// encodings + id cursors per graph). See [`MOON_AUX_VECTOR_DEFS`]. pub const MOON_AUX_GRAPH_STORE: &[u8] = b"moon-graph-store"; +/// Moon replication aux key: the whole process-global `WorkspaceRegistry` +/// (`replication::ws_sync::export_workspace_registry` blob). Unlike the other +/// `MOON_AUX_*` blobs this one is NOT sharded — exactly one entry exists in +/// 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"; /// `write_rdb_refs` plus moon-private AUX fields written immediately after /// the standard header aux block (before any SELECTDB), which is what lets diff --git a/src/replication/mod.rs b/src/replication/mod.rs index b6a7e18b..8e906d50 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -7,3 +7,4 @@ pub mod master; pub mod reason_del; pub mod replica; pub mod state; +pub mod ws_sync; diff --git a/src/replication/ws_sync.rs b/src/replication/ws_sync.rs new file mode 100644 index 00000000..09c8eaac --- /dev/null +++ b/src/replication/ws_sync.rs @@ -0,0 +1,195 @@ +//! v0.7 WS-plane PSYNC snapshot: export / install the process-global +//! `WorkspaceRegistry` as a moon-private RDB aux blob (Wave B ws-plane). +//! +//! Unlike vector/text/graph state, the workspace registry is NOT sharded — it +//! is one `Mutex>>` on `ShardDatabases`, shared +//! by every shard's thread (`shard/shared_databases.rs`). Only shard 0 +//! captures it (shard-0-authoritative, mirroring the WS.CREATE/DROP +//! write-path pin — the registry's replication-offset accounting lives +//! entirely on shard 0's thread, so the snapshot leg captures it from the +//! same place): the single-shard master path +//! (`master::handle_psync_inline_single_shard`) captures it directly; the +//! multi-shard merge path (`master::handle_psync_inline_multi_shard`) +//! captures it ONLY in shard 0's `PrepareReplicaSync` leg +//! (`PreparedShardSync::ws_registry_blob`), leaving it `None` on every other +//! shard so the merge loop's "keep the first `Some`" convention (same as +//! `vector_defs`/`text_defs`) picks shard 0's copy. +//! +//! Blob format (version 1, little-endian): +//! ```text +//! [u8 version = 1] +//! [u32 entry_count] +//! per entry: +//! [16 bytes ws_id] +//! [u32 name_len][name bytes] +//! [i64 created_at_ms] +//! ``` +//! +//! An empty registry (or no registry ever initialized) still encodes as +//! `entry_count = 0` — this lets the replica distinguish "master +//! authoritatively has zero workspaces" (install an empty registry, +//! replacing whatever local entries exist) from "pre-WS-sync master, aux +//! absent entirely" (warn and keep local state), the same convention +//! `export_graph_store` uses for graphs. + +use bytes::Bytes; + +use crate::workspace::WorkspaceId; +use crate::workspace::registry::{WorkspaceMetadata, WorkspaceRegistry}; + +const FORMAT_VERSION: u8 = 1; + +/// Export the registry as a snapshot blob. `registry: None` (never +/// initialized) encodes identically to `Some(empty)`. +pub fn export_workspace_registry(registry: Option<&WorkspaceRegistry>) -> Vec { + let count = registry.map_or(0, |r| r.len()); + let mut buf = Vec::with_capacity(5 + count * 32); + buf.push(FORMAT_VERSION); + buf.extend_from_slice(&(count as u32).to_le_bytes()); + if let Some(reg) = registry { + for (id, meta) in reg.iter() { + buf.extend_from_slice(id.as_bytes()); + buf.extend_from_slice(&(meta.name.len() as u32).to_le_bytes()); + buf.extend_from_slice(&meta.name); + buf.extend_from_slice(&meta.created_at.to_le_bytes()); + } + } + buf +} + +/// Decode a snapshot blob into a fresh `WorkspaceRegistry` (authoritative +/// replace — the caller installs it in place of whatever the replica already +/// had). Returns `None` on a malformed blob (truncated / unknown version) — +/// the caller aborts the sync and the replica retries with a fresh full +/// resync, matching `graph_sync::install_graph_store`'s contract. +pub fn install_workspace_registry(blob: &[u8]) -> Option { + let mut cur = Cursor { data: blob, pos: 0 }; + if cur.u8()? != FORMAT_VERSION { + return None; + } + let count = cur.u32()? as usize; + let mut reg = WorkspaceRegistry::new(); + for _ in 0..count { + let mut id_bytes = [0u8; 16]; + id_bytes.copy_from_slice(cur.take(16)?); + let name_len = cur.u32()? as usize; + let name = Bytes::copy_from_slice(cur.take(name_len)?); + let created_at = cur.i64()?; + let id = WorkspaceId::from_bytes(id_bytes); + reg.insert( + id, + WorkspaceMetadata { + id, + name, + created_at, + }, + ); + } + Some(reg) +} + +/// Minimal bounds-checked reader over the blob (mirrors `graph_sync::Cursor`). +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 i64(&mut self) -> Option { + #[allow(clippy::unwrap_used)] // take(8) guarantees the length + Some(i64::from_le_bytes(self.take(8)?.try_into().unwrap())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_meta(id: WorkspaceId, name: &str, created_at: i64) -> WorkspaceMetadata { + WorkspaceMetadata { + id, + name: Bytes::from(name.to_owned()), + created_at, + } + } + + #[test] + fn export_install_round_trip_preserves_entries() { + let mut reg = WorkspaceRegistry::new(); + let id1 = WorkspaceId::from_bytes([1u8; 16]); + let id2 = WorkspaceId::from_bytes([2u8; 16]); + reg.insert(id1, make_meta(id1, "acme", 1_752_000_000_000)); + reg.insert(id2, make_meta(id2, "widgets", -5)); + + let blob = export_workspace_registry(Some(®)); + let installed = install_workspace_registry(&blob).expect("valid blob"); + + assert_eq!(installed.len(), 2); + let m1 = installed.get(&id1).expect("id1 present"); + assert_eq!(m1.name.as_ref(), b"acme"); + assert_eq!(m1.created_at, 1_752_000_000_000); + let m2 = installed.get(&id2).expect("id2 present"); + assert_eq!(m2.name.as_ref(), b"widgets"); + assert_eq!(m2.created_at, -5); + } + + #[test] + fn none_and_empty_registry_both_encode_as_zero_entries() { + let empty = WorkspaceRegistry::new(); + assert_eq!( + export_workspace_registry(None), + export_workspace_registry(Some(&empty)) + ); + let installed = install_workspace_registry(&export_workspace_registry(None)).unwrap(); + assert!(installed.is_empty()); + } + + #[test] + fn malformed_blob_rejected() { + assert!(install_workspace_registry(b"").is_none()); + assert!(install_workspace_registry(&[9u8]).is_none()); // bad version, no count + assert!(install_workspace_registry(&[FORMAT_VERSION]).is_none()); // truncated count + + // Truncated mid-entry: valid header claiming 1 entry but no data. + let mut bad = vec![FORMAT_VERSION]; + bad.extend_from_slice(&1u32.to_le_bytes()); + assert!(install_workspace_registry(&bad).is_none()); + } + + #[test] + fn truncated_full_blob_rejected() { + let mut reg = WorkspaceRegistry::new(); + let id = WorkspaceId::from_bytes([3u8; 16]); + reg.insert(id, make_meta(id, "acme", 42)); + let blob = export_workspace_registry(Some(®)); + assert!(install_workspace_registry(&blob[..blob.len() - 3]).is_none()); + } + + #[test] + fn empty_name_and_negative_created_at_survive() { + let mut reg = WorkspaceRegistry::new(); + let id = WorkspaceId::from_bytes([4u8; 16]); + reg.insert(id, make_meta(id, "", -1)); + let blob = export_workspace_registry(Some(®)); + let installed = install_workspace_registry(&blob).unwrap(); + let m = installed.get(&id).unwrap(); + assert!(m.name.is_empty()); + assert_eq!(m.created_at, -1); + } +} From b23c1dd21578b4027a20d6e751895fafcb1b00c7 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:51:46 +0700 Subject: [PATCH 3/9] feat(shard): WS registry shard-0 affinity hop (WsRegistryCreate/Drop) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane. Adds the cross-shard hop that will let any connection thread route the process-global WorkspaceRegistry mutation onto shard 0's own OS thread, mirroring the existing `WsDropCleanup` hop shape. New `ShardMessage::WsRegistryCreate(Box)` / `WsRegistryDrop` variants (boxed vs inline sized per the existing 64-byte enum budget) and their `handle_shard_message_shared` arms run the WHOLE mutation + WAL-append + replication-record sequence in one synchronous stretch on shard 0's thread — this is what keeps the replication offset advance atomic with the mutation w.r.t. a concurrent PSYNC snapshot capture, the same argument `record_local_write` relies on for per-shard state, applied here to the one piece of state that is global. A new `record_ws_registry_write` helper duplicates (rather than shares) `record_local_write`'s backlog-append + offset-advance + self-queue-push steps, because this call site has no `ConnectionContext` — only the raw per-shard pieces already threaded through `handle_shard_message_shared`'s signature (same precedent as `wal_append_and_fanout` being an independent, not shared, copy of the same steps). Also wires the WS registry export into `PrepareReplicaSync`'s reply (`PreparedShardSync::ws_registry_blob`, `Some` only on shard 0's leg, `None` elsewhere) so the master-side PSYNC capture sites can pick it up in a follow-up commit. Not yet reachable from any command handler — `WsRegistryCreate`/`Drop` are constructed nowhere yet; that lands with the write-path wiring. author: Tin Dang --- src/shard/dispatch.rs | 45 ++++++++++++++ src/shard/spsc_handler.rs | 122 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 167 insertions(+) diff --git a/src/shard/dispatch.rs b/src/shard/dispatch.rs index 61086728..a427b5c6 100644 --- a/src/shard/dispatch.rs +++ b/src/shard/dispatch.rs @@ -492,6 +492,17 @@ pub struct PrepareReplicaSyncPayload { pub reply_tx: channel::MpscSender, } +/// Payload of [`ShardMessage::WsRegistryCreate`] (Wave B ws-plane). +pub struct WsRegistryCreatePayload { + /// The MASTER-minted workspace id (UUIDv7, allocated on the connecting + /// thread BEFORE the hop) — shard 0 applies this verbatim, never + /// allocating its own. + pub ws_id: [u8; 16], + pub name: Bytes, + pub created_at: i64, + pub reply_tx: channel::OneshotSender<()>, +} + /// One shard's prepared full-resync leg — the reply to /// [`ShardMessage::PrepareReplicaSync`]. pub struct PreparedShardSync { @@ -514,6 +525,12 @@ pub struct PreparedShardSync { /// replica imports all of them (`read_moon_aux_all`). #[cfg(feature = "graph")] pub graph_blob: Vec, + /// The process-global `WorkspaceRegistry` snapshot blob + /// (`replication::ws_sync::export_workspace_registry`). The registry is + /// NOT sharded, so only shard 0's leg populates this (`Some`); every + /// 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>, } /// Messages sent to a shard via SPSC channels from the connection layer @@ -823,6 +840,34 @@ pub enum ShardMessage { prefix: Bytes, reply_tx: channel::OneshotSender, }, + /// WS.CREATE shard-0 affinity hop (Wave B ws-plane). + /// + /// The workspace registry is process-global, so its mutation + WAL + /// record + replication live-push must run in ONE synchronous stretch on + /// shard 0's own thread — otherwise the replication offset advance + /// (which must happen on shard 0's thread; see `record_local_write`'s + /// snapshot-consistency argument) could race a concurrent PSYNC snapshot + /// capture on shard 0. A connection whose OWN shard is 0 runs the + /// sequence inline (no hop, already on that thread); every other shard's + /// connection sends this message to shard 0 and awaits the reply — same + /// dual-path shape as `WsDropCleanup`, except the "owner" here is always + /// shard 0, never a hash-tag-derived shard. + /// + /// Boxed: `[u8;16](16) + Bytes(32) + i64(8) + OneshotSender(8)` = 64 B, + /// which would push the enum past the 64-byte inline cap once the + /// discriminant and alignment padding are added. + WsRegistryCreate(Box), + /// WS.DROP shard-0 affinity hop (Wave B ws-plane) — same rationale as + /// [`ShardMessage::WsRegistryCreate`]. Reply carries whether the id + /// existed (mirrors the connection handler's pre-hop "removed: bool" + /// check); the caller's SEPARATE `WsDropCleanup` key-sweep hop (unchanged + /// by this variant) only runs when this reply is `true`. + /// + /// Inline: `[u8;16](16) + OneshotSender(8)` = 24 B — within the cap. + WsRegistryDrop { + ws_id: [u8; 16], + reply_tx: channel::OneshotSender, + }, /// AOF cooperative-snapshot hop (C2/C4, shardslice-migration Wave A1). /// /// The AOF rewrite writer pushes this message into the shard's external diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs index dcbb0630..f9115d3a 100644 --- a/src/shard/spsc_handler.rs +++ b/src/shard/spsc_handler.rs @@ -2579,6 +2579,76 @@ pub(crate) fn handle_shard_message_shared( // completed; losing the ack is harmless. let _ = reply_tx.send(deleted_count); } + ShardMessage::WsRegistryCreate(payload) => { + // Wave B ws-plane: the WHOLE sequence (global-registry insert, + // WorkspaceCreate WAL record, replication live-push) runs here in + // ONE synchronous stretch on shard 0's own thread — this is what + // keeps the replication offset advance atomic with the mutation + // w.r.t. a concurrent PSYNC snapshot capture (same argument as + // `record_local_write`, applied to the one piece of state that is + // global rather than per-shard). `shard_id` is 0 here by + // construction: only shard 0 ever receives this message (see + // `try_handle_ws_command`'s hop routing). + let crate::shard::dispatch::WsRegistryCreatePayload { + ws_id, + name, + created_at, + reply_tx, + } = *payload; + let id = crate::workspace::WorkspaceId::from_bytes(ws_id); + { + let mut guard = shard_databases.workspace_registry(); + let reg = guard + .get_or_insert_with(|| Box::new(crate::workspace::WorkspaceRegistry::new())); + reg.insert( + id, + crate::workspace::registry::WorkspaceMetadata { + id, + name: name.clone(), + created_at, + }, + ); + } + let wal_payload = + crate::workspace::wal::encode_workspace_create(&ws_id, &name, created_at); + shard_databases.wal_append( + shard_id, + crate::persistence::wal_v3::record::WalRecordType::WorkspaceCreate, + bytes::Bytes::from(wal_payload.clone()), + ); + if crate::replication::state::fanout_hint_active() { + let record = crate::workspace::repl::serialize_ws_create_apply(&wal_payload); + record_ws_registry_write(repl_backlog, repl_state, shard_id, record.into()); + } + let _ = reply_tx.send(()); + } + ShardMessage::WsRegistryDrop { ws_id, reply_tx } => { + // Wave B ws-plane: mirrors `WsRegistryCreate` above. `removed` + // mirrors the pre-hop connection-thread check — the caller's + // SEPARATE `WsDropCleanup` key-sweep hop only fires when this is + // `true`. + let id = crate::workspace::WorkspaceId::from_bytes(ws_id); + let removed = { + let mut guard = shard_databases.workspace_registry(); + match guard.as_mut() { + Some(reg) => reg.remove(&id).is_some(), + None => false, + } + }; + if removed { + let wal_payload = crate::workspace::wal::encode_workspace_drop(&ws_id); + shard_databases.wal_append( + shard_id, + crate::persistence::wal_v3::record::WalRecordType::WorkspaceDrop, + bytes::Bytes::from(wal_payload.clone()), + ); + if crate::replication::state::fanout_hint_active() { + let record = crate::workspace::repl::serialize_ws_drop_apply(&wal_payload); + record_ws_registry_write(repl_backlog, repl_state, shard_id, record.into()); + } + } + let _ = reply_tx.send(removed); + } ShardMessage::AofFold { reply_tx } => { // AOF cooperative-snapshot (C4): build an expired-filtered snapshot // of ALL databases on this shard and reply. The AOF rewrite writer @@ -2819,6 +2889,20 @@ pub(crate) fn handle_shard_message_shared( crate::replication::graph_sync::export_graph_store(&mut s.graph_store); } }); + // Wave B ws-plane: the workspace registry is process-global, not + // per-shard, so only shard 0 (the sole writer of it — see + // `try_handle_ws_command`'s shard-0 hop) captures it. Capturing + // it here keeps it in the SAME synchronous stretch as this + // shard's offset read below, mirroring the master's-thread + // atomicity argument the WS write path relies on. + let ws_registry_blob = if shard_id == 0 { + let guard = shard_databases.workspace_registry(); + Some(crate::replication::ws_sync::export_workspace_registry( + guard.as_deref(), + )) + } else { + None + }; let shard_offset = repl_state .as_ref() .map(|h| h.shard_offset(shard_id)) @@ -2849,6 +2933,7 @@ pub(crate) fn handle_shard_message_shared( text_defs, #[cfg(feature = "graph")] graph_blob, + ws_registry_blob, }; if reply_tx.try_send(prepared).is_err() { // The PSYNC task is gone (replica dropped mid-handshake) — @@ -3718,6 +3803,43 @@ pub(crate) fn fanout_send_or_kick( }); } +/// Push a WS.CREATE.APPLY / WS.DROP.APPLY replication record (Wave B +/// ws-plane): backlog append + offset advance happen HERE, synchronously +/// (same argument as `record_local_write`/`wal_append_and_fanout`'s steps +/// 2-4), and the live replica delivery is deferred through the shard's +/// self-queue exactly like every other replication record. +/// +/// A standalone helper rather than a call into `handler_monoio::ft`'s +/// `record_local_write`/`record_local_write_db`: this runs inside +/// `handle_shard_message_shared` (the `ShardMessage::WsRegistryCreate` / +/// `WsRegistryDrop` arms), which has no `ConnectionContext` — only the raw +/// per-shard pieces (`repl_backlog`, `repl_state`, `shard_id`) that are +/// already threaded through this function's signature. WS records are +/// db-agnostic (applied to the global registry, not a `Database`), so unlike +/// `wal_append_and_fanout` there is no `SELECT` prefix to thread — the apply +/// side ignores db context for these commands. +fn record_ws_registry_write( + repl_backlog: &crate::replication::backlog::SharedBacklog, + repl_state: &Option, + shard_id: usize, + bytes: bytes::Bytes, +) { + let mut end_offset = u64::MAX; + if let Some(offsets) = repl_state { + { + let mut guard = repl_backlog.lock(); + if let Some(backlog) = guard.as_mut() { + backlog.append(&bytes); + } + } + end_offset = offsets.increment_shard_offset(shard_id, bytes.len() as u64); + } + crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { + bytes, + end_offset, + }); +} + /// Error frame substituted for a write's success frame when the command /// mutated memory but its AOF record could not be enqueued within the /// backpressure budget — fail-loud so the client knows durability was not From 432633e458419cf67fff1bc620954d61bb3cd0a0 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:52:04 +0700 Subject: [PATCH 4/9] feat(replication): WS.CREATE/DROP replica apply arms + snapshot install MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane. Threads `Arc` through the replica's apply path so it can reach the process-global `WorkspaceRegistry`, which lives on `ShardDatabases` rather than the thread-local `ShardSlice`: - `apply::apply_local` recognizes the `WS.CREATE.APPLY` / `WS.DROP.APPLY` internal pseudo-commands (`workspace::repl`) before generic dispatch. `apply_ws_create` installs the MASTER's `ws_id` + `created_at` VERBATIM — UUIDv7 is nondeterministic, so a replica must never mint its own id. `apply_ws_drop` removes the entry and reruns the master's best-effort `{ws_hex}:`-prefix key sweep across every db on this shard (R0 replication is single-shard-only, so sweeping the one shard IS the whole sweep — no cross-shard hop to mirror). Both are in-memory only, matching `apply_graph`'s no-local-WAL-persistence model: a restarted replica resyncs the registry from its master rather than replaying a local copy. - `apply::load_snapshot` reads the (singular, shard-0-authoritative) `MOON_AUX_WORKSPACE_REGISTRY` aux blob and installs it authoritatively after a successful keyspace load — same convention as the graph store (empty blob = master authoritatively has none, replaces local state; absent aux = pre-WS-sync master, warn-and-keep-local). - `ReplicaTaskConfig` gains a `shard_databases` field (both runtime variants' 4 `apply_local`/`load_snapshot` call sites updated); its two construction sites (`handler_monoio`/`handler_sharded` `dispatch.rs`) now pass `ctx.shard_databases.clone()`. Not yet reachable — the master-side PSYNC capture and the write-path shard-0 hop wiring land in the next two commits. author: Tin Dang --- src/replication/apply.rs | 154 +++++++++++++++++++- src/replication/replica.rs | 13 +- src/server/conn/handler_monoio/dispatch.rs | 1 + src/server/conn/handler_sharded/dispatch.rs | 1 + 4 files changed, 162 insertions(+), 7 deletions(-) diff --git a/src/replication/apply.rs b/src/replication/apply.rs index 364a4b53..c747c36a 100644 --- a/src/replication/apply.rs +++ b/src/replication/apply.rs @@ -149,13 +149,33 @@ fn frame_to_usize(f: &Frame) -> Option { /// Returns `false` only if this thread has no initialized `ShardSlice` — which /// would indicate the replica task was spawned off a shard thread (a wiring /// bug); the caller logs and drops the stream. -pub(crate) fn apply_local(rc: &ReplCommand) -> bool { +/// +/// `shard_databases` gives the WS.CREATE.APPLY / WS.DROP.APPLY arms +/// (Wave B ws-plane) access to the process-global `WorkspaceRegistry`, which +/// lives on `ShardDatabases` rather than the thread-local `ShardSlice` — R0 +/// replication is single-shard-only, so applying directly here (no shard-0 +/// hop) is correct: there is exactly one shard. +pub(crate) fn apply_local( + rc: &ReplCommand, + shard_databases: &std::sync::Arc, +) -> bool { use crate::command::{DispatchResult, dispatch as cmd_dispatch}; use crate::shard::spsc_handler::extract_command_static; let Some((cmd, args)) = extract_command_static(&rc.command) else { return true; // not an array command — nothing to apply (defensive) }; + if cmd.eq_ignore_ascii_case(crate::workspace::repl::WS_CREATE_APPLY_CMD) { + // Doesn't touch `ShardSlice`, but routes through `try_with_shard` + // anyway so a replica task wired to a non-shard thread is caught + // uniformly (same "no ShardSlice here" contract every other arm has). + return crate::shard::slice::try_with_shard(|_s| apply_ws_create(shard_databases, args)) + .is_some(); + } + if cmd.eq_ignore_ascii_case(crate::workspace::repl::WS_DROP_APPLY_CMD) { + return crate::shard::slice::try_with_shard(|s| apply_ws_drop(s, shard_databases, args)) + .is_some(); + } crate::shard::slice::try_with_shard(|s| { let db_count = s.databases.len(); if db_count == 0 { @@ -246,6 +266,85 @@ pub(crate) fn apply_local(rc: &ReplCommand) -> bool { .is_some() } +/// Apply a replicated `WS.CREATE.APPLY` record (Wave B ws-plane): install the +/// MASTER's `ws_id` + `name` + `created_at` VERBATIM into the process-global +/// workspace registry. UUIDv7 is nondeterministic — a replica must never mint +/// its own; it applies exactly what the master already decided (round-2 +/// finding A / task #34). In-memory only, matching `apply_graph`'s +/// no-local-persistence model — a restarted replica resyncs from its master +/// rather than replaying a local WAL copy of this record. +fn apply_ws_create( + shard_databases: &std::sync::Arc, + args: &[Frame], +) { + let Some(payload) = crate::workspace::repl::extract_payload(args) else { + tracing::warn!("replica apply: WS.CREATE.APPLY missing payload — skipped"); + return; + }; + let Some((ws_id_bytes, name, created_at)) = + crate::workspace::wal::decode_workspace_create(&payload) + else { + tracing::warn!("replica apply: WS.CREATE.APPLY payload malformed — skipped"); + return; + }; + let id = crate::workspace::WorkspaceId::from_bytes(ws_id_bytes); + let mut guard = shard_databases.workspace_registry(); + let reg = guard.get_or_insert_with(|| Box::new(crate::workspace::WorkspaceRegistry::new())); + reg.insert( + id, + crate::workspace::registry::WorkspaceMetadata { + id, + name: bytes::Bytes::from(name), + created_at, + }, + ); +} + +/// Apply a replicated `WS.DROP.APPLY` record (Wave B ws-plane): remove the +/// workspace from the registry, then run the SAME best-effort key-prefix +/// sweep the master's local-owner branch of `WsDropCleanup` runs (R0 +/// replication is single-shard-only, so there is no cross-shard hop to +/// mirror — sweeping every db on this one shard IS the whole sweep). +fn apply_ws_drop( + s: &mut crate::shard::slice::ShardSlice, + shard_databases: &std::sync::Arc, + args: &[Frame], +) { + let Some(payload) = crate::workspace::repl::extract_payload(args) else { + tracing::warn!("replica apply: WS.DROP.APPLY missing payload — skipped"); + return; + }; + let Some(ws_id_bytes) = crate::workspace::wal::decode_workspace_drop(&payload) else { + tracing::warn!("replica apply: WS.DROP.APPLY payload malformed — skipped"); + return; + }; + let id = crate::workspace::WorkspaceId::from_bytes(ws_id_bytes); + let removed = { + let mut guard = shard_databases.workspace_registry(); + match guard.as_mut() { + Some(reg) => reg.remove(&id).is_some(), + None => false, + } + }; + if !removed { + // Already absent (e.g. a re-delivered record after a resync) — + // nothing to sweep. + return; + } + let prefix = format!("{{{}}}:", id.as_hex()); + let prefix_bytes = prefix.into_bytes(); + for db in s.databases.iter_mut() { + let keys_to_delete: Vec> = db + .keys() + .filter(|k| k.as_bytes().starts_with(&prefix_bytes[..])) + .map(|k| k.as_bytes().to_vec()) + .collect(); + for key in &keys_to_delete { + db.remove(key); + } + } +} + /// Apply a replicated FT.* index-definition command (FT.CREATE / FT.DROPINDEX /// / FT.CONFIG SET) through the same handlers the master's connection layer /// uses. Deliberately NO keyspace backfill on live FT.CREATE — master parity: @@ -456,7 +555,15 @@ fn apply_two_db( /// /// Returns the number of keys loaded, or an error if this thread has no /// `ShardSlice` or the RDB is malformed. -pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { +/// +/// `shard_databases` carries the process-global `WorkspaceRegistry` +/// (Wave B ws-plane) — installed AFTER the keyspace load, same authoritative- +/// replace semantics as the graph store below, but outside `try_with_shard` +/// since the registry lives on `ShardDatabases`, not `ShardSlice`. +pub(crate) fn load_snapshot( + rdb: &[u8], + shard_databases: &std::sync::Arc, +) -> anyhow::Result { use crate::persistence::redis_rdb; // Moon-private aux fields (written by the master right after the RDB // header) carry the FT index DEFINITIONS; standard RDB loaders skip them. @@ -467,7 +574,10 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { // them all; a single-shard snapshot yields exactly one. #[cfg(feature = "graph")] let graph_blobs = redis_rdb::read_moon_aux_all(rdb, redis_rdb::MOON_AUX_GRAPH_STORE); - match crate::shard::slice::try_with_shard(|s| { + // 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); + let result: anyhow::Result = match crate::shard::slice::try_with_shard(|s| { for db in s.databases.iter_mut() { db.clear(); } @@ -516,7 +626,45 @@ pub(crate) fn load_snapshot(rdb: &[u8]) -> anyhow::Result { None => Err(anyhow::anyhow!( "replica snapshot load: no ShardSlice on this thread" )), + }; + let loaded = result?; + + // Wave B ws-plane: install the master's workspace registry AFTER a + // successful keyspace load — authoritative replace (an EMPTY blob drops + // replica-local workspaces; an ABSENT aux means a pre-WS-sync master, + // warn-and-keep), same convention as the graph store above. + match ws_registry_blob { + Some(blob) => match crate::replication::ws_sync::install_workspace_registry(&blob) { + Some(reg) => { + let count = reg.len(); + *shard_databases.workspace_registry() = Some(Box::new(reg)); + if count > 0 { + tracing::info!("replica snapshot: installed {} workspace(s)", count); + } + } + None => { + return Err(anyhow::anyhow!( + "replica snapshot: malformed workspace-registry aux blob" + )); + } + }, + None => { + let existing = shard_databases + .workspace_registry() + .as_ref() + .map_or(0, |r| r.len()); + if existing > 0 { + tracing::warn!( + "replica snapshot carried no workspace-registry aux but {} local \ + workspace(s) exist — master predates WS replication; keeping local \ + registry (it may diverge)", + existing + ); + } + } } + + Ok(loaded) } /// Full resync = authoritative replace: drop every replica-local FT index, diff --git a/src/replication/replica.rs b/src/replication/replica.rs index 4167a405..7b0b0970 100644 --- a/src/replication/replica.rs +++ b/src/replication/replica.rs @@ -17,6 +17,7 @@ use tracing::{info, warn}; use crate::replication::handshake::ReplicaHandshakeState; use crate::replication::state::{ReplicationRole, ReplicationState, save_replication_state}; +use crate::shard::shared_databases::ShardDatabases; /// Process-global generation counter for replica tasks (attach-under-write /// P0, found while testing R2): `REPLICAOF host port` used to spawn a fresh @@ -55,6 +56,10 @@ pub struct ReplicaTaskConfig { pub num_shards: usize, pub persistence_dir: Option, pub listening_port: u16, + /// Gives `apply::load_snapshot` / `apply::apply_local` access to the + /// process-global `WorkspaceRegistry` (Wave B ws-plane), which lives on + /// `ShardDatabases` rather than the thread-local `ShardSlice`. + pub shard_databases: Arc, /// Generation ticket from [`bump_replica_task_epoch`] — the task exits /// as soon as a newer generation exists. pub epoch: u64, @@ -267,7 +272,7 @@ async fn run_handshake_and_stream( } for shard_id in 0..cfg.num_shards { let rdb_bytes = read_rdb_bulk(&mut stream).await?; - match crate::replication::apply::load_snapshot(&rdb_bytes) { + match crate::replication::apply::load_snapshot(&rdb_bytes, &cfg.shard_databases) { Ok(keys) => info!( "Replica: loaded shard {} RDB snapshot ({} bytes, {} keys)", shard_id, @@ -380,7 +385,7 @@ async fn stream_commands_read_loop( let outcome = crate::replication::apply::drain_replicated_commands(&mut buf, &mut selected_db); for rc in &outcome.commands { - if !crate::replication::apply::apply_local(rc) { + if !crate::replication::apply::apply_local(rc, &cfg.shard_databases) { return Err(anyhow::anyhow!( "replica has no ShardSlice on this thread — cannot apply replication stream" )); @@ -598,7 +603,7 @@ async fn run_handshake_and_stream( } for shard_id in 0..cfg.num_shards { let rdb_bytes = read_rdb_bulk(&mut stream).await?; - match crate::replication::apply::load_snapshot(&rdb_bytes) { + match crate::replication::apply::load_snapshot(&rdb_bytes, &cfg.shard_databases) { Ok(keys) => info!( "Replica: loaded shard {} RDB snapshot ({} bytes, {} keys)", shard_id, @@ -720,7 +725,7 @@ async fn stream_commands_read_loop( let outcome = crate::replication::apply::drain_replicated_commands(&mut buf, &mut selected_db); for rc in &outcome.commands { - if !crate::replication::apply::apply_local(rc) { + if !crate::replication::apply::apply_local(rc, &cfg.shard_databases) { return Err(anyhow::anyhow!( "replica has no ShardSlice on this thread — cannot apply replication stream" )); diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 079843b6..24be81ba 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -478,6 +478,7 @@ pub(super) fn try_handle_replicaof( listening_port: 0, epoch, stream_db: std::sync::atomic::AtomicUsize::new(0), + shard_databases: ctx.shard_databases.clone(), }; monoio::spawn(crate::replication::replica::run_replica_task(cfg)); } diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 3bc32319..f939311b 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -295,6 +295,7 @@ pub(super) fn try_handle_replicaof( listening_port: 0, epoch, stream_db: std::sync::atomic::AtomicUsize::new(0), + shard_databases: ctx.shard_databases.clone(), }; tokio::task::spawn_local(crate::replication::replica::run_replica_task(cfg)); } From d8bb33ad2fb6e9f8d9adc396211ff519d6b80c39 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:52:32 +0700 Subject: [PATCH 5/9] feat(replication): capture WS registry blob at both master PSYNC sites Wave B ws-plane. Wires `replication::ws_sync::export_workspace_registry` into the two FULLRESYNC capture sites: - `handle_psync_inline_single_shard`: shard 0 IS this thread's shard, so the capture is trivially in the same synchronous stretch as the offset read; always written to `moon_aux` (empty blob tells the replica the master authoritatively has zero workspaces), same convention as the graph blob. - `handle_psync_inline_multi_shard`: only shard 0's `PreparedShardSync` leg populates `ws_registry_blob` (every other shard replies `None`, wired in the previous commit); the merge loop keeps the first `Some`, matching the `vector_defs`/`text_defs` convention. Combined with the previous commit's `load_snapshot` install, a replica attaching to a master that already has workspaces now backfills them from the snapshot leg. author: Tin Dang --- src/replication/master.rs | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/src/replication/master.rs b/src/replication/master.rs index 9b5d9649..c85f8280 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -625,7 +625,7 @@ pub async fn handle_psync_inline_single_shard( client_offset: i64, mut stream: monoio::net::TcpStream, repl_state: Arc>, - _shard_databases: Arc, + shard_databases: Arc, replica_addr: std::net::SocketAddr, ) -> anyhow::Result<()> { use monoio::io::AsyncWriteRentExt; @@ -739,6 +739,15 @@ pub async fn handle_psync_inline_single_shard( #[cfg(feature = "graph")] let graph_blob = crate::replication::graph_sync::export_graph_store(&mut s.graph_store); + // Wave B ws-plane: the workspace registry snapshot. Shard + // 0 IS this thread's shard (single-shard path), so this + // capture is trivially in the same synchronous stretch as + // the offset read above — same convention as the graph + // blob: always written, an empty blob (0 entries) tells + // the replica the master authoritatively has none. + let ws_registry_blob = crate::replication::ws_sync::export_workspace_registry( + shard_databases.workspace_registry().as_deref(), + ); let mut moon_aux: Vec<(&[u8], &[u8])> = Vec::new(); if let Some(ref v) = vec_defs { moon_aux @@ -752,6 +761,10 @@ pub async fn handle_psync_inline_single_shard( crate::persistence::redis_rdb::MOON_AUX_GRAPH_STORE, &graph_blob[..], )); + moon_aux.push(( + crate::persistence::redis_rdb::MOON_AUX_WORKSPACE_REGISTRY, + &ws_registry_blob[..], + )); crate::persistence::redis_rdb::write_rdb_refs_with_moon_aux( &refs, &moon_aux, @@ -941,6 +954,11 @@ pub async fn handle_psync_inline_multi_shard( let mut snapshot_offset: u64 = 0; #[cfg(feature = "graph")] let mut graph_blobs: Vec> = Vec::with_capacity(num_shards); + // Wave B ws-plane: the registry is process-global, so only shard 0's leg + // populates this (`Some`) — every other shard replies `None` (see + // `PreparedShardSync::ws_registry_blob`). "Keep the first Some" matches + // the `vector_defs`/`text_defs` convention below. + let mut ws_registry_blob: Option> = None; 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 @@ -979,6 +997,9 @@ pub async fn handle_psync_inline_multi_shard( if text_defs.is_none() { text_defs = prepared.text_defs; } + if ws_registry_blob.is_none() { + ws_registry_blob = prepared.ws_registry_blob; + } #[cfg(feature = "graph")] graph_blobs.push(prepared.graph_blob); bodies.push(prepared.rdb_body); @@ -1000,6 +1021,12 @@ pub async fn handle_psync_inline_multi_shard( &blob[..], )); } + if let Some(w) = &ws_registry_blob { + moon_aux.push(( + crate::persistence::redis_rdb::MOON_AUX_WORKSPACE_REGISTRY, + &w[..], + )); + } let mut rdb_buf: Vec = Vec::new(); crate::persistence::redis_rdb::write_rdb_merged(&moon_aux, &bodies, &mut rdb_buf); info!( From 62cf584d3fc7ac8715e19522ab541c21ec6cd4a3 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:52:48 +0700 Subject: [PATCH 6/9] feat(command): route WS.CREATE/DROP through the shard-0 affinity hop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane — the write-path wiring that makes everything from the previous five commits reachable. `try_handle_ws_command` (monoio) now branches on `ctx.shard_id == 0`: a connection already on shard 0 runs the mutation + WAL-append + replication- record sequence inline (as before, just now also emitting the live replication record); every other connection sends `WsRegistryCreate` / `WsRegistryDrop` to shard 0 via `spsc_send` and awaits the bounded oneshot reply — same dual-path shape `WsDropCleanup` already used for the key-sweep half of WS.DROP, now applied to the registry mutation itself. Command semantics (return value, error cases) are unchanged; only the mutation's execution thread and its now-replicated side effect are new. Live replication emission happens only on the inline (shard-0) branch, gated on `replication_fanout_active(ctx)` exactly like the graph plane: the WorkspaceCreate/WorkspaceDrop WAL payload is wrapped via `workspace::repl::serialize_ws_create_apply`/`serialize_ws_drop_apply` and pushed with `record_local_write_db`. The foreign-hop branch's equivalent push happens on shard 0's own thread inside `handle_shard_message_shared` (landed in commit 3) — so every WS.CREATE/DROP is replicated exactly once regardless of which connection thread received it. Also splits the former combined `warn_unreplicated_plane` fail-loud marker (`handler_monoio/ft.rs`) into `warn_mq_unreplicated`: the WS half is deleted now that this plane replicates; the MQ half survives (MQ.* is not yet wired into replication — separate branch/task). author: Tin Dang --- src/server/conn/handler_monoio/ft.rs | 27 ++-- src/server/conn/handler_monoio/write.rs | 167 +++++++++++++++++------- 2 files changed, 134 insertions(+), 60 deletions(-) diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index 8b5b9aca..447d596c 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/src/server/conn/handler_monoio/ft.rs @@ -164,21 +164,28 @@ fn serialize_select(db: usize) -> Vec { buf } -/// Fail-loud marker for planes NOT yet wired into replication (round-2 -/// finding A): WS.* and MQ.* writes persist durably on the master but never -/// reach a replica — deterministic record forms + replica apply arms are the -/// task-#34 follow-up. 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. -pub(super) fn warn_unreplicated_plane(ctx: &ConnectionContext, cmd: &[u8]) { +/// 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: WS.*/MQ.* writes are NOT replicated in v0.7 — a \ - replica will not see this plane (known limitation, further \ - occurrences not logged)" + "replication: MQ.* writes are NOT replicated in v0.7 — a replica \ + will not see this plane (known limitation, further occurrences \ + not logged)" ); } } diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 48456161..09366d4a 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -46,16 +46,6 @@ pub(super) async fn try_handle_ws_command( } }; - // Round-2 finding A fail-loud: WS.CREATE/WS.DROP persist locally - // (WorkspaceCreate/Drop WAL records) but are NOT replicated in v0.7 — - // surface the divergence once instead of letting a replica silently miss - // the plane. (WS.CREATE is non-deterministic — fresh UUIDv7 per execution - // — so verbatim streaming would be wrong; task #34 tracks the id-pinned - // record form.) - if sub.eq_ignore_ascii_case(b"CREATE") || sub.eq_ignore_ascii_case(b"DROP") { - super::ft::warn_unreplicated_plane(ctx, cmd); - } - if sub.eq_ignore_ascii_case(b"CREATE") { match validate_ws_create(cmd_args) { Ok(ws_name) => { @@ -64,33 +54,76 @@ pub(super) async fn try_handle_ws_command( .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as i64; - let meta = crate::workspace::WorkspaceMetadata { - id: ws_id, - name: ws_name.clone(), - created_at, - }; - { - let mut guard = ctx.shard_databases.workspace_registry(); - let reg = guard.get_or_insert_with(|| { - Box::new(crate::workspace::WorkspaceRegistry::new()) - }); - reg.insert(ws_id, meta); + // Wave B ws-plane: the registry is process-global, so the + // WHOLE mutation+WAL+replication-record sequence pins to + // shard 0's own thread — the replication offset advance must + // stay in the SAME synchronous stretch as the mutation (see + // `record_local_write`'s snapshot-consistency argument). A + // connection whose OWN shard is 0 runs the sequence inline + // (no hop, already on that thread); every other connection + // sends `WsRegistryCreate` to shard 0 and awaits the reply — + // same dual-path shape as `WsDropCleanup` below, except the + // "owner" here is always shard 0. + if ctx.shard_id == 0 { + let meta = crate::workspace::WorkspaceMetadata { + id: ws_id, + name: ws_name.clone(), + created_at, + }; + { + let mut guard = ctx.shard_databases.workspace_registry(); + let reg = guard.get_or_insert_with(|| { + Box::new(crate::workspace::WorkspaceRegistry::new()) + }); + reg.insert(ws_id, meta); + } + // WAL: WorkspaceCreate record (unframed, real type — K1a). + // K1b: the payload carries `created_at` so it survives a + // restart. + let wal_payload = crate::workspace::wal::encode_workspace_create( + ws_id.as_bytes(), + &ws_name, + created_at, + ); + ctx.shard_databases.wal_append( + 0, + crate::persistence::wal_v3::record::WalRecordType::WorkspaceCreate, + Bytes::from(wal_payload.clone()), + ); + // Wave B ws-plane: live replication push (same payload as + // the WAL record, wrapped as the internal + // WS.CREATE.APPLY pseudo-command — see + // `workspace::repl`). Gated on fanout-active exactly like + // the graph plane's `record_local_write_db` call. + if super::ft::replication_fanout_active(ctx) { + let record = + crate::workspace::repl::serialize_ws_create_apply(&wal_payload); + super::ft::record_local_write_db( + ctx, + conn.selected_db, + Bytes::from(record), + ); + } + } else { + let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); + let msg = crate::shard::dispatch::ShardMessage::WsRegistryCreate(Box::new( + crate::shard::dispatch::WsRegistryCreatePayload { + ws_id: *ws_id.as_bytes(), + name: ws_name.clone(), + created_at, + reply_tx, + }, + )); + let _ = crate::shard::coordinator::spsc_send( + &ctx.dispatch_tx, + ctx.shard_id, + 0, + msg, + &ctx.spsc_notifiers, + ) + .await; + let _ = crate::shard::coordinator::recv_reply_bounded(reply_rx).await; } - // WAL: WorkspaceCreate record (unframed, real type — K1a). The - // registry is global, so its WAL stream is pinned to shard 0 — - // one stream gives replay a total order over Create/Drop - // regardless of which connection issued them. K1b: the payload - // carries `created_at` so it survives a restart. - let payload = crate::workspace::wal::encode_workspace_create( - ws_id.as_bytes(), - &ws_name, - created_at, - ); - ctx.shard_databases.wal_append( - 0, - crate::persistence::wal_v3::record::WalRecordType::WorkspaceCreate, - Bytes::from(payload), - ); responses.push(Frame::BulkString(Bytes::from(ws_id.to_string()))); } Err(e) => responses.push(e), @@ -103,22 +136,56 @@ pub(super) async fn try_handle_ws_command( Ok(ws_id_raw) => { match parse_workspace_id_from_bytes(&ws_id_raw) { Some(ws_id) => { - let removed = { - let mut guard = ctx.shard_databases.workspace_registry(); - match guard.as_mut() { - Some(reg) => reg.remove(&ws_id).is_some(), - None => false, + // Wave B ws-plane: same shard-0 hop as WS.CREATE above. + let removed = if ctx.shard_id == 0 { + let removed = { + let mut guard = ctx.shard_databases.workspace_registry(); + match guard.as_mut() { + Some(reg) => reg.remove(&ws_id).is_some(), + None => false, + } + }; + if removed { + // WAL: WorkspaceDrop record (unframed, real type — K1a). + let wal_payload = + crate::workspace::wal::encode_workspace_drop(ws_id.as_bytes()); + ctx.shard_databases.wal_append( + 0, + crate::persistence::wal_v3::record::WalRecordType::WorkspaceDrop, + Bytes::from(wal_payload.clone()), + ); + if super::ft::replication_fanout_active(ctx) { + let record = crate::workspace::repl::serialize_ws_drop_apply( + &wal_payload, + ); + super::ft::record_local_write_db( + ctx, + conn.selected_db, + Bytes::from(record), + ); + } } + removed + } else { + let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); + let msg = crate::shard::dispatch::ShardMessage::WsRegistryDrop { + ws_id: *ws_id.as_bytes(), + reply_tx, + }; + let _ = crate::shard::coordinator::spsc_send( + &ctx.dispatch_tx, + ctx.shard_id, + 0, + msg, + &ctx.spsc_notifiers, + ) + .await; + matches!( + crate::shard::coordinator::recv_reply_bounded(reply_rx).await, + Ok(true) + ) }; if removed { - // WAL: WorkspaceDrop record (unframed, real type — K1a). - let payload = - crate::workspace::wal::encode_workspace_drop(ws_id.as_bytes()); - ctx.shard_databases.wal_append( - 0, - crate::persistence::wal_v3::record::WalRecordType::WorkspaceDrop, - Bytes::from(payload), - ); // Best-effort cleanup: delete all KV keys with ws prefix (WS-03). // Owner-route the cleanup via WsDropCleanup hop. The {wsid} hash // tag co-locates every workspace key on ONE shard. @@ -313,7 +380,7 @@ pub(super) async fn try_handle_mq_command( // 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_unreplicated_plane(ctx, cmd); + super::ft::warn_mq_unreplicated(ctx, cmd); } if sub.eq_ignore_ascii_case(b"CREATE") { From 7751775e8e9f0c48a82dc6bbce47f7697818b21b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 11:53:02 +0700 Subject: [PATCH 7/9] test(replication): WS.CREATE/DROP replication coverage + CHANGELOG MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave B ws-plane. New `tests/replication_ws.rs`, self-contained (own copies of `start_moon`/`Killer`/`send_cmd`/`send_resp`/`wait_until`, modeled on `tests/replication_graph.rs`): - `replica_syncs_ws_create_live_stream` — replica attaches to an EMPTY master, then WS.CREATE runs; asserts the replica resolves the MASTER's exact `ws_id` via WS.INFO/WS.AUTH (the verbatim-apply / no-re-mint correctness guard). - `replica_backfills_ws_registry_from_snapshot` — WS.CREATE runs BEFORE the replica ever attaches; asserts FULLRESYNC backfill produces an identical WS.INFO on both sides, and that a live write past the snapshot boundary still replicates. - `replica_syncs_ws_drop` — WS.DROP with a replica attached; asserts the workspace disappears from WS.LIST and WS.AUTH fails identically on both sides. - `ws_create_shard0_hop_consistent_across_shards` — `--shards 4`, no replica: WS.CREATE followed by 16 connections' WS.INFO/WS.LIST, exercising the shard-0 hop from whichever shard SO_REUSEPORT lands each connection on. Also adds the `[Unreleased]` CHANGELOG entry summarizing the six-commit Wave B ws-plane series. Verified: all 4 new tests green against a `release-fast` build; existing `replication_graph` (3), `replication_planes` (11), and `shardslice_live` (6, including both `test_ssm3_ws_*` legs) stay green — no regressions. Both default and `runtime-tokio,jemalloc` clippy matrices clean at `-D warnings`; `cargo fmt --check` clean; full `cargo test --profile release-fast --lib` green (4156 passed, 1 ignored, 0 failed). author: Tin Dang --- CHANGELOG.md | 41 +++++ tests/replication_ws.rs | 371 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 412 insertions(+) create mode 100644 tests/replication_ws.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4325ad5b..77802002 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,47 @@ binary — WS/MQ are only wired in the sharded handlers, so the synthetic single-shard harness used by `tests/replication_test.rs` can't exercise this fix). Verified against both a monoio (default-feature) build and a `runtime-tokio,jemalloc` build. +### Added — WS.CREATE/DROP replication (Wave B ws-plane) + +`WS.CREATE`/`WS.DROP` mutated the process-global `WorkspaceRegistry` from +whichever connection thread received the command and pinned their WAL +records to shard 0 without actually running on shard 0's thread — the +replication offset advance (which must happen synchronously with the +mutation on shard 0, matching the snapshot-capture atomicity argument used +elsewhere) had no such guarantee, and the plane was not replicated at all +(round-2 finding A / task #34). + +Routed the whole mutation + WAL-append + replication-record sequence through +a dedicated shard-0 hop (`ShardMessage::WsRegistryCreate` / +`WsRegistryDrop`, mirroring the existing `WsDropCleanup` hop): a connection +already on shard 0 runs the sequence inline, every other connection hops +over via `spsc_send` + a bounded oneshot reply. After the shard-0 execution, +the WorkspaceCreate/WorkspaceDrop record (same payload as the WAL record — +`ws_id` + `name` + `created_at_ms`) is pushed into the replication stream via +the graph-plane's `record_local_write_db` pattern (offset advances IFF the +backlog append happens). + +Replica apply arms (`WS.CREATE.APPLY` / `WS.DROP.APPLY` internal +pseudo-commands, `replication::apply::apply_local`) install the MASTER's +`ws_id` + `created_at` VERBATIM — UUIDv7 is nondeterministic, so a replica +must never mint its own id. `WS.DROP.APPLY` reruns the master's best-effort +`{ws_hex}:`-prefix key sweep on the replica's single shard (R0 replication is +single-shard-only). Full-resync backfill: a new versioned +`MOON_AUX_WORKSPACE_REGISTRY` RDB aux blob (`replication::ws_sync`, +shard-0-authoritative — captured only in shard 0's `PrepareReplicaSync` leg +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. + +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. ### Fixed — docs site strict build red since 2026-07-08 (root-relative links) diff --git a/tests/replication_ws.rs b/tests/replication_ws.rs new file mode 100644 index 00000000..eae64799 --- /dev/null +++ b/tests/replication_ws.rs @@ -0,0 +1,371 @@ +//! Wave B ws-plane replication: a replica must synchronize the process-global +//! `WorkspaceRegistry` (WS.CREATE/WS.DROP) from its master — both the live +//! mutation stream and the PSYNC full-resync snapshot backfill. +//! +//! WS.CREATE mints a nondeterministic UUIDv7 on the master; the replica must +//! apply the MASTER's `ws_id` verbatim (never mint its own), so these tests +//! assert on the EXACT id (and `created_at`) round-tripping, not just +//! presence. Modeled on `tests/replication_graph.rs`. +//! +//! Run: `MOON_BIN=./target/release/moon cargo test --test replication_ws -- --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) -> Child { + start_moon_shards(port, dir, 1) +} + +fn start_moon_shards(port: u16, dir: &str, shards: u32) -> Child { + 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(Stdio::null()) + .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 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() +} + +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() +} + +/// Extract the workspace id from a `WS CREATE` bulk-string reply +/// (`$\r\n\r\n`). +fn extract_ws_id(create_reply: &str) -> String { + create_reply + .lines() + .nth(1) + .unwrap_or_default() + .trim() + .to_string() +} + +/// REPL-WS-01: live WS.CREATE on the master reaches the replica — including +/// the EXACT `ws_id` (nondeterministic UUIDv7, never re-minted by the +/// replica) and `created_at`. The replica attaches to an EMPTY master first, +/// isolating the live streaming leg from snapshot backfill. +#[test] +#[ignore] +fn replica_syncs_ws_create_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 _master = Killer(start_moon(16740, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + let _replica = Killer(start_moon(16741, replica_dir.path().to_str().unwrap())); + 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" + ); + std::thread::sleep(Duration::from_millis(500)); + + let create_reply = send_resp(master_addr, &["WS", "CREATE", "acme"]); + assert!( + create_reply.starts_with('$'), + "WS CREATE must return a bulk-string id: {create_reply}" + ); + let ws_id = extract_ws_id(&create_reply); + assert!(!ws_id.is_empty(), "WS CREATE returned an empty id"); + + let synced = wait_until(Duration::from_secs(10), || { + send_cmd(replica_addr, "WS LIST").contains("acme") + }); + assert!( + synced, + "replica never saw the live-created workspace: {}", + send_cmd(replica_addr, "WS LIST") + ); + + // The EXACT master-minted id must round-trip — WS.INFO on the replica + // must resolve it (a re-minted id on the replica would 404 here). + let info = send_resp(replica_addr, &["WS", "INFO", &ws_id]); + assert!( + info.contains("acme"), + "replica WS INFO for master's exact ws_id failed: {info}" + ); + + // WS AUTH against the master's exact id must also succeed on the replica + // (the id the client already has from the master's CREATE response). + let auth = send_resp(replica_addr, &["WS", "AUTH", &ws_id]); + assert!( + auth.starts_with("+OK"), + "replica WS AUTH with master's exact ws_id failed: {auth}" + ); +} + +/// REPL-WS-02: a replica attaching to a master that ALREADY has a workspace +/// must backfill it from the PSYNC full-resync snapshot (not just live +/// deltas) — including the exact `ws_id` and `created_at`. +#[test] +#[ignore] +fn replica_backfills_ws_registry_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())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + + // Build WS state on the master BEFORE any replica exists. + let create_reply = send_resp(master_addr, &["WS", "CREATE", "presnap"]); + assert!(create_reply.starts_with('$'), "seed WS CREATE failed"); + let ws_id = extract_ws_id(&create_reply); + let master_info = send_resp(master_addr, &["WS", "INFO", &ws_id]); + + // Now attach a fresh replica — it must learn `presnap` + its exact id. + let _replica = Killer(start_moon(16743, replica_dir.path().to_str().unwrap())); + 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), || { + send_cmd(replica_addr, "WS LIST").contains("presnap") + }); + assert!( + backfilled, + "replica never backfilled the pre-existing workspace: {}", + send_cmd(replica_addr, "WS LIST") + ); + + let replica_info = send_resp(replica_addr, &["WS", "INFO", &ws_id]); + assert_eq!( + replica_info, master_info, + "replica WS INFO diverged from the master's snapshot-backfilled entry" + ); + + // A live write AFTER the snapshot must also land (stream continues past + // the backfill boundary). + let create2 = send_resp(master_addr, &["WS", "CREATE", "postsnap"]); + assert!(create2.starts_with('$'), "post-snapshot WS CREATE failed"); + let grew = wait_until(Duration::from_secs(10), || { + send_cmd(replica_addr, "WS LIST").contains("postsnap") + }); + assert!(grew, "post-snapshot live WS.CREATE did not replicate"); +} + +/// REPL-WS-03: WS.DROP propagates to the replica — the workspace disappears +/// from WS.LIST and a subsequent WS.AUTH against the dropped id fails on the +/// replica exactly as it does on the master. +#[test] +#[ignore] +fn replica_syncs_ws_drop() { + 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())); + 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())); + 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)); + + let create_reply = send_resp(master_addr, &["WS", "CREATE", "todrop"]); + assert!(create_reply.starts_with('$'), "WS CREATE failed"); + let ws_id = extract_ws_id(&create_reply); + + let synced = wait_until(Duration::from_secs(10), || { + send_cmd(replica_addr, "WS LIST").contains("todrop") + }); + assert!(synced, "replica never saw the pre-drop workspace"); + + // Bind a connection to the workspace on the master and write a + // workspace-scoped key, so DROP's key-sweep has something to clean up on + // both sides (mirrors the master's WsDropCleanup best-effort sweep). + let mut bound = TcpStream::connect(master_addr).expect("connect bound conn"); + bound + .set_read_timeout(Some(Duration::from_millis(500))) + .ok(); + let auth_cmd = format!( + "*3\r\n$2\r\nWS\r\n$4\r\nAUTH\r\n${}\r\n{}\r\n", + ws_id.len(), + ws_id + ); + bound.write_all(auth_cmd.as_bytes()).unwrap(); + read_quiet(&mut bound); + bound + .write_all(b"*3\r\n$3\r\nSET\r\n$4\r\nwkey\r\n$3\r\nval\r\n") + .unwrap(); + read_quiet(&mut bound); + drop(bound); + + let drop_reply = send_resp(master_addr, &["WS", "DROP", &ws_id]); + assert!( + drop_reply.starts_with("+OK"), + "WS DROP failed: {drop_reply}" + ); + + let gone = wait_until(Duration::from_secs(10), || { + !send_cmd(replica_addr, "WS LIST").contains("todrop") + }); + assert!( + gone, + "replica never applied WS.DROP: {}", + send_cmd(replica_addr, "WS LIST") + ); + + // WS AUTH against the dropped id must now fail identically on both sides. + let master_auth = send_resp(master_addr, &["WS", "AUTH", &ws_id]); + let replica_auth = send_resp(replica_addr, &["WS", "AUTH", &ws_id]); + assert!( + master_auth.starts_with('-') && replica_auth.starts_with('-'), + "post-drop WS AUTH must error on both sides: master={master_auth:?} replica={replica_auth:?}" + ); +} + +/// REPL-WS-04: the shard-0 affinity hop itself (no replica involved) — at +/// `--shards 4`, WS.CREATE issued from ANY connection (the kernel spreads +/// SO_REUSEPORT-accepted connections across shard listeners; on Linux this +/// routinely lands non-zero shards, exercising the cross-shard hop to shard +/// 0) must be immediately, consistently visible from every other connection. +/// Mirrors `tests/shardslice_live.rs`'s `test_ssm3_ws_global_registry` oracle +/// but additionally asserts `created_at` consistency across connections +/// (Wave B moved the write off the connecting thread onto shard 0's). +#[test] +#[ignore] +fn ws_create_shard0_hop_consistent_across_shards() { + let dir = tempfile::tempdir().unwrap(); + let addr = "127.0.0.1:16746"; + let _server = Killer(start_moon_shards(16746, dir.path().to_str().unwrap(), 4)); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(addr, "PING") + .starts_with("+PONG")), + "server never became ready" + ); + + let create_reply = send_resp(addr, &["WS", "CREATE", "hop4"]); + assert!(create_reply.starts_with('$'), "WS CREATE failed"); + let ws_id = extract_ws_id(&create_reply); + let baseline_info = send_resp(addr, &["WS", "INFO", &ws_id]); + + for i in 0..16 { + let info = send_resp(addr, &["WS", "INFO", &ws_id]); + assert_eq!( + info, baseline_info, + "conn {i}: WS INFO diverged across connections (shard-0 hop inconsistency)" + ); + let list_flat = send_cmd(addr, "WS LIST"); + assert!( + list_flat.contains("hop4"), + "conn {i}: WS LIST must include 'hop4' from every shard connection" + ); + } +} From 3ace974b8e579e49efbe21fd11a74b4087d54e53 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 12:10:27 +0700 Subject: [PATCH 8/9] fix(workspace): fail-closed WS.CREATE hop + WS.DROP timeout sweep recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the ws-plane branch (verdict SHIP-WITH-FIXES): P0: the WS.CREATE shard-0 hop discarded both the spsc_send and recv_reply_bounded outcomes and unconditionally replied with the ws_id — a wedged ring or 30s reply timeout returned a phantom id the registry never saw (WS AUTH/INFO against it 404s), directly contradicting the sibling WS.DROP path that already gated on the reply. Fixed: the hop outcome is checked, and on failure the process-global registry is probed to disambiguate "shard 0 executed but the reply was slow/lost" (entry present, success) from "the mutation never happened" (entry absent, fail closed with a retryable error). P1: a WS.DROP hop timeout resolved `removed = false` even when shard 0 completed the drop (timeout means "no reply in 30s", not "failed" — the handler's synchronous O(keys × databases) WsDropCleanup scan makes slow replies plausible). The registry entry being gone meant no retry could ever re-trigger the {ws_hex}: key-prefix sweep — the workspace's keys leaked permanently. Fixed with the same registry probe: entry gone → the drop landed, run the (idempotent) sweep; entry present → genuine hop failure, fall through to the not-found path as before. Both fixes are conn-thread-only; the shard-0 handler and replication paths are untouched. Gates re-run green: lib 4156/4156, replication_ws 4/4 (incl. the 16-iteration cross-shard hop consistency test), clippy both matrices, fmt. author: Tin Dang --- src/server/conn/handler_monoio/write.rs | 47 +++++++++++++++++++++---- 1 file changed, 41 insertions(+), 6 deletions(-) diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index 09366d4a..e5c3047a 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/src/server/conn/handler_monoio/write.rs @@ -104,6 +104,7 @@ pub(super) async fn try_handle_ws_command( Bytes::from(record), ); } + responses.push(Frame::BulkString(Bytes::from(ws_id.to_string()))); } else { let (reply_tx, reply_rx) = crate::runtime::channel::oneshot(); let msg = crate::shard::dispatch::ShardMessage::WsRegistryCreate(Box::new( @@ -122,9 +123,28 @@ pub(super) async fn try_handle_ws_command( &ctx.spsc_notifiers, ) .await; - let _ = crate::shard::coordinator::recv_reply_bounded(reply_rx).await; + // Fail CLOSED on hop failure (adversarial-review P0): a + // wedged ring or reply timeout must not report a phantom + // ws_id the registry never saw. The registry itself is + // process-global, so a post-failure probe disambiguates + // "shard 0 executed but the reply was slow/lost" (entry + // present → success) from "the mutation never happened" + // (entry absent → retryable error). + let hop_ok = crate::shard::coordinator::recv_reply_bounded(reply_rx) + .await + .is_ok(); + let created = hop_ok || { + let guard = ctx.shard_databases.workspace_registry(); + guard.as_ref().is_some_and(|reg| reg.get(&ws_id).is_some()) + }; + if created { + responses.push(Frame::BulkString(Bytes::from(ws_id.to_string()))); + } else { + responses.push(Frame::Error(Bytes::from_static( + b"ERR workspace create failed (shard-0 registry hop timed out); not created, safe to retry", + ))); + } } - responses.push(Frame::BulkString(Bytes::from(ws_id.to_string()))); } Err(e) => responses.push(e), } @@ -180,10 +200,25 @@ pub(super) async fn try_handle_ws_command( &ctx.spsc_notifiers, ) .await; - matches!( - crate::shard::coordinator::recv_reply_bounded(reply_rx).await, - Ok(true) - ) + match crate::shard::coordinator::recv_reply_bounded(reply_rx).await { + Ok(removed) => removed, + // Timeout/err ≠ failure (adversarial-review + // P1): shard 0 may have completed the drop + // and only the REPLY was lost — resolving + // `false` here skipped the `{ws_hex}:` key + // sweep below FOREVER (the registry entry is + // gone, so no WS.DROP retry can re-trigger + // it). The registry is process-global: + // probe it — entry gone → the drop landed + // (run the sweep; it is idempotent), entry + // still present → the hop genuinely never + // executed → fall through to + // ERR_WS_NOT_FOUND-side handling as before. + Err(_) => { + let guard = ctx.shard_databases.workspace_registry(); + guard.as_ref().is_none_or(|reg| reg.get(&ws_id).is_none()) + } + } }; if removed { // Best-effort cleanup: delete all KV keys with ws prefix (WS-03). From fc023ac53c64c17f9c0a6d9f610f3480083ca361 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Sun, 12 Jul 2026 12:11:51 +0700 Subject: [PATCH 9/9] ci(fuzz): register ws_registry_record in PR + nightly fuzz matrices The ws-plane branch added the fuzz target binary but not its CI matrix entries (Testing rule: every new decoder fuzz target runs in CI). author: Tin Dang --- .github/workflows/fuzz.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index e1c7a3a5..305140d1 100644 --- a/.github/workflows/fuzz.yml +++ b/.github/workflows/fuzz.yml @@ -36,6 +36,7 @@ jobs: - graph_props_record - fts_query_parse - mq_wal_record + - ws_registry_record steps: - uses: actions/checkout@v7 - uses: dtolnay/rust-toolchain@nightly @@ -88,6 +89,7 @@ jobs: - graph_props_record - fts_query_parse - mq_wal_record + - ws_registry_record steps: - uses: actions/checkout@v7 - uses: dtolnay/rust-toolchain@nightly