feat(replication): WS plane — shard-0-pinned deterministic WS.CREATE/DROP + live streaming + PSYNC registry blob (Wave 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

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)
 

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

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

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

src/replication/apply.rs

@@ -149,13 +149,33 @@ fn frame_to_usize(f: &Frame) -> Option<usize> {
 /// 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<crate::shard::shared_databases::ShardDatabases>,
+) -> 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<crate::shard::shared_databases::ShardDatabases>,
+    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<crate::shard::shared_databases::ShardDatabases>,
+    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<Vec<u8>> = 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<usize> {
+///
+/// `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<crate::shard::shared_databases::ShardDatabases>,
+) -> anyhow::Result<usize> {
     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<usize> {
     // 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<usize> = 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<usize> {
         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,

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<RwLock<ReplicationState>>,
-    _shard_databases: Arc<crate::shard::shared_databases::ShardDatabases>,
+    shard_databases: Arc<crate::shard::shared_databases::ShardDatabases>,
     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<u8>> = 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<Vec<u8>> = 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<u8> = Vec::new();
     crate::persistence::redis_rdb::write_rdb_merged(&moon_aux, &bodies, &mut rdb_buf);
     info!(

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;