feat(shard): ShardSlice Phase 4 — shared-nothing storage live, lock wrappers deleted

.add/state.json

@@ -1,7 +1,7 @@
 {
   "project": "moon",
   "stage": "production",
-  "active_task": "consistency-dispatch-gaps",
+  "active_task": "shardslice-migration",
   "active_milestone": "v1-shared-nothing",
   "tasks": {
     "hotpath-lock-quickwins": {
@@ -33,6 +33,21 @@
       "created": "2026-06-11T10:38:27+00:00",
       "updated": "2026-06-12T02:21:52+00:00",
       "flag_verified": true
+    },
+    "shardslice-migration": {
+      "title": "Wire ShardSlice: thread-local shared-nothing storage becomes the live path",
+      "phase": "done",
+      "gate": "RISK-ACCEPTED",
+      "milestone": "v1-shared-nothing",
+      "depends_on": [],
+      "created": "2026-06-12T03:23:36+00:00",
+      "updated": "2026-06-12T17:04:31+00:00",
+      "flag_verified": true,
+      "waiver": {
+        "owner": "Tin Dang",
+        "ticket": "follow-up task: cross-shard-read-acceleration (observe)",
+        "expires": "2026-08-01"
+      }
     }
   },
   "milestones": {
@@ -46,7 +61,7 @@
     }
   },
   "created": "2026-06-11T03:18:21+00:00",
-  "updated": "2026-06-12T02:21:52+00:00",
+  "updated": "2026-06-12T17:04:31+00:00",
   "setup": {
     "locked": true,
     "locked_at": "2026-06-11T03:28:00+00:00",

src/admin/metrics_setup.rs

@@ -1391,41 +1391,31 @@ pub fn spawn_moon_memory_publisher() {
 /// Collect per-subsystem resident bytes and emit all 7
 /// `moon_memory_bytes{kind=...}` series plus `moon_rss_bytes`.
 ///
-/// Called every 15 s by `spawn_moon_memory_publisher`. Allocation-free in
-/// the steady state — all label values are `&'static str`.
+/// Called every 15 s by `spawn_moon_memory_publisher`. Lock-free: reads from
+/// per-shard published atomics (C5 / M4). Figures lag at most one 100ms tick.
 fn update_moon_memory_bytes() {
+    use std::sync::atomic::Ordering;
+
     let rss = get_rss_bytes() as usize;
 
     let mut dashtable: usize = 0;
     let mut hnsw: usize = 0;
-    let mut sealed: usize = 0;
-    #[cfg_attr(not(feature = "graph"), allow(unused_mut))]
+    let sealed: usize = 0; // combined into hnsw from vector atomic (C5)
     let mut csr: usize = 0;
     let wal: usize = 0; // WalWriterV3 is stack-owned; not reachable here
     let mut backlog: usize = 0;
 
     if let Some(shard_dbs) = get_global_shard_databases() {
-        let num_shards = shard_dbs.num_shards();
-        for shard_id in 0..num_shards {
-            // Database + DashTable (DB 0 — the hot database)
-            let db_guard = shard_dbs.read_db(shard_id, 0);
-            dashtable += db_guard.resident_bytes();
-            dashtable += db_guard.data().resident_bytes();
-            drop(db_guard);
-
-            // VectorStore: (mutable/hnsw, immutable/sealed)
-            let vs = shard_dbs.vector_store(shard_id);
-            let (m, i) = vs.resident_bytes();
-            hnsw += m;
-            sealed += i;
-            drop(vs);
-
-            // GraphStore (CSR)
-            #[cfg(feature = "graph")]
-            {
-                let gs = shard_dbs.graph_store_read(shard_id);
-                csr += gs.resident_bytes();
-            }
+        // KV memory: sum of per-shard published atomics. Lock-free.
+        // C5 / M4: `read_memory_sum()` replaces per-shard `read_db(…)` locks.
+        dashtable = shard_dbs.read_memory_sum();
+
+        // Store memory: sum published per-shard vector/graph atomics.
+        // Values are refreshed by each shard's 100ms tick (publish_store_memory).
+        for mem in shard_dbs.store_memory_per_shard.iter() {
+            hnsw += mem.vector.load(Ordering::Relaxed);
+            // graph is cfg-gated at publish time; the atomic is always present.
+            csr += mem.graph.load(Ordering::Relaxed);
         }
     }
 

src/command/persistence.rs

@@ -441,7 +441,7 @@ mod tests {
         // Wrap as a TopLevel pool to match the post-2e-β helper signature.
         let (tx, _rx) = crate::runtime::channel::mpsc_bounded::<AofMessage>(1);
         let pool = AofWriterPool::top_level(tx);
-        let shard_dbs = crate::shard::shared_databases::ShardDatabases::new(vec![vec![
+        let (shard_dbs, _inits) = crate::shard::shared_databases::ShardDatabases::new(vec![vec![
             crate::storage::Database::new(),
         ]]);
 
@@ -505,7 +505,7 @@ mod tests {
         let _guard = GATE_TEST_LOCK.lock();
         let (tx, _rx) = crate::runtime::channel::mpsc_bounded::<AofMessage>(1);
         let pool = AofWriterPool::top_level(tx);
-        let shard_dbs = crate::shard::shared_databases::ShardDatabases::new(vec![vec![
+        let (shard_dbs, _inits) = crate::shard::shared_databases::ShardDatabases::new(vec![vec![
             crate::storage::Database::new(),
         ]]);
 

src/command/server_admin.rs

@@ -383,37 +383,34 @@ fn memory_doctor() -> Frame {
     let rss = crate::admin::metrics_setup::get_rss_bytes() as usize;
     let vsz = get_vsz_bytes();
 
-    // ── Gather per-subsystem resident bytes ──────────────────────────────
-    let mut dashtable_bytes: usize = 0;
-    let mut hnsw_bytes: usize = 0;
-    let mut sealed_bytes: usize = 0;
-    #[cfg_attr(not(feature = "graph"), allow(unused_mut))]
-    let mut csr_bytes: usize = 0;
+    // ── Gather per-subsystem resident bytes (C5 / M4 — lock-free atomics) ──
+    // KV and store memory are read from published per-shard atomics. Figures
+    // lag at most one 100ms tick — acceptable for an on-demand diagnostic.
+    use std::sync::atomic::Ordering;
+    let dashtable_bytes: usize;
+    let hnsw_bytes: usize;
+    let sealed_bytes: usize = 0; // combined into hnsw_bytes from vector atomic
+    #[cfg_attr(not(feature = "graph"), allow(unused_variables))]
+    let csr_bytes: usize;
     let wal_bytes: usize = 0;
 
     if let Some(shard_dbs) = crate::admin::metrics_setup::get_global_shard_databases() {
-        let num_shards = shard_dbs.num_shards();
-        for shard_id in 0..num_shards {
-            // Database + DashTable (DB 0 only — the hot database)
-            let db_guard = shard_dbs.read_db(shard_id, 0);
-            dashtable_bytes += db_guard.resident_bytes();
-            dashtable_bytes += db_guard.data().resident_bytes();
-            drop(db_guard);
-
-            // VectorStore: (mutable/hnsw, immutable/sealed)
-            let vs = shard_dbs.vector_store(shard_id);
-            let (mutable, immutable) = vs.resident_bytes();
-            hnsw_bytes += mutable;
-            sealed_bytes += immutable;
-            drop(vs);
-
-            // GraphStore (CSR)
-            #[cfg(feature = "graph")]
-            {
-                let gs = shard_dbs.graph_store_read(shard_id);
-                csr_bytes += gs.resident_bytes();
-            }
+        // KV memory: sum of per-shard published atomics. Lock-free.
+        dashtable_bytes = shard_dbs.read_memory_sum();
+
+        // Store memory: sum published per-shard vector/graph atomics.
+        let mut vec_total = 0usize;
+        let mut csr_total = 0usize;
+        for mem in shard_dbs.store_memory_per_shard.iter() {
+            vec_total += mem.vector.load(Ordering::Relaxed);
+            csr_total += mem.graph.load(Ordering::Relaxed);
         }
+        hnsw_bytes = vec_total;
+        csr_bytes = csr_total;
+    } else {
+        dashtable_bytes = 0;
+        hnsw_bytes = 0;
+        csr_bytes = 0;
     }
 
     // Replication backlog via global state (same pattern as INFO replication).

src/main.rs

@@ -59,7 +59,7 @@ use moon::runtime::channel;
 use moon::runtime::{RuntimeFactoryImpl, traits::RuntimeFactory};
 use moon::server;
 use moon::shard::Shard;
-use moon::shard::mesh::{CHANNEL_BUFFER_SIZE, ChannelMesh};
+use moon::shard::mesh::{CHANNEL_BUFFER_SIZE, ChannelMesh, create_aof_fold_channels};
 use moon::shard::shared_databases::ShardDatabases;
 use tracing::info;
 
@@ -601,7 +601,7 @@ fn main() -> anyhow::Result<()> {
         std::process::exit(2);
     }
 
-    let aof_pool: Option<std::sync::Arc<AofWriterPool>> = if config.appendonly == "yes" {
+    let mut aof_pool: Option<std::sync::Arc<AofWriterPool>> = if config.appendonly == "yes" {
         let fsync = FsyncPolicy::from_str(&config.appendfsync);
         // PerShard writers required when num_shards >= 2 AND we'll have a
         // PerShard manifest at runtime. Two cases produce PerShard:
@@ -787,6 +787,39 @@ fn main() -> anyhow::Result<()> {
     // Collect all notifiers before spawning shard threads
     let all_notifiers = mesh.all_notifiers();
 
+    // C4: Wire AOF fold channels for the per-shard cooperative snapshot protocol.
+    //
+    // The pool is built BEFORE the mesh (line ~650) so fold channels cannot be
+    // set at construction time — we set them here via set_fold_channels, using
+    // Arc::get_mut which is safe at this point because no shard threads have
+    // spawned yet (the first Arc clone happens at ~line 1290).
+    //
+    // fold_consumers[i] is merged into shard i's consumers vec before
+    // shard.run() is called, mirroring the admin_consumers merge pattern.
+    //
+    // Only PerShard pools participate in the per-shard rewrite; TopLevel pools
+    // (--shards 1 or legacy single-writer) do not need fold channels.
+    let mut fold_consumers: Option<Vec<ringbuf::HeapCons<moon::shard::dispatch::ShardMessage>>> =
+        None;
+    if let Some(ref mut pool_arc) = aof_pool
+        && pool_arc.layout() == moon::persistence::aof_manifest::AofLayout::PerShard
+    {
+        let (fold_producers, fold_cons) = create_aof_fold_channels(num_shards, 4);
+        // SAFETY: Arc::get_mut is valid here — no clones exist yet; the first
+        // shard_aof_pool clone happens inside the shard-spawn loop below.
+        if let Some(pool_mut) = std::sync::Arc::get_mut(pool_arc) {
+            pool_mut.set_fold_channels(fold_producers, all_notifiers.clone());
+            fold_consumers = Some(fold_cons);
+        } else {
+            // Should never happen at this point in startup.
+            tracing::error!(
+                "C4 wiring: Arc::get_mut failed — fold channels not wired. \
+                 Per-shard BGREWRITEAOF will abort cleanly (old generation \
+                 remains authoritative) rather than deadlock."
+            );
+        }
+    }
+
     // Create admin SPSC channels for the console gateway (one per shard).
     #[cfg(feature = "console")]
     let mut admin_consumers = {
@@ -1227,32 +1260,37 @@ fn main() -> anyhow::Result<()> {
         .iter_mut()
         .map(|s| std::mem::take(&mut s.databases))
         .collect();
-    let shard_databases = ShardDatabases::new(all_dbs);
+    let (shard_databases, mut slice_inits) = ShardDatabases::new(all_dbs);
 
     // Recover graph stores from persistence (CSR segments + metadata + WAL replay).
+    // These run on ShardSliceInit (mutate pre-shard state before handoff to threads).
     #[cfg(feature = "graph")]
     if let Some(ref dir) = persistence_dir {
         let dir_path = std::path::Path::new(dir);
-        shard_databases.recover_graph_stores(dir_path);
-        shard_databases.replay_graph_wal(dir_path);
+        moon::shard::shared_databases::recover_graph_stores(&mut slice_inits, dir_path);
+        moon::shard::shared_databases::replay_graph_wal(
+            &mut slice_inits,
+            dir_path,
+            config.databases,
+        );
     }
 
     // Replay temporal WAL records (not gated on graph feature — temporal KV is core).
     if let Some(ref dir) = persistence_dir {
         let dir_path = std::path::Path::new(dir);
-        shard_databases.replay_temporal_wal(dir_path);
+        moon::shard::shared_databases::replay_temporal_wal(&mut slice_inits, dir_path);
     }
 
-    // Replay workspace WAL records (not gated on graph feature — workspaces are core).
+    // Replay workspace WAL records — uses shared registry, takes Arc<ShardDatabases>.
     if let Some(ref dir) = persistence_dir {
         let dir_path = std::path::Path::new(dir);
-        shard_databases.replay_workspace_wal(dir_path);
+        moon::shard::shared_databases::replay_workspace_wal(&shard_databases, dir_path);
     }
 
     // Replay MQ WAL records (cursor-rollback for durable queues).
     if let Some(ref dir) = persistence_dir {
         let dir_path = std::path::Path::new(dir);
-        shard_databases.replay_mq_wal(dir_path);
+        moon::shard::shared_databases::replay_mq_wal(&mut slice_inits, dir_path);
     }
 
     // All shards recovered — mark server as ready for /readyz.
@@ -1280,6 +1318,28 @@ fn main() -> anyhow::Result<()> {
             });
             consumers.push(admin_cons);
         }
+        // C4: Prepend AOF fold consumer for this shard (aof-writer-N -> shard SPSC).
+        // The consumer receives ShardMessage::AofFold pushed by the per-shard AOF
+        // writer thread; the shard event loop processes it between commands and
+        // replies with a frozen snapshot. Only wired for PerShard pools.
+        //
+        // PRIORITY: the fold consumer is inserted at index 0 (before command
+        // consumers) so that AofFold is never starved by MAX_DRAIN_PER_CYCLE.
+        // Under sustained write load the command consumer can saturate the
+        // 256-message drain cap every cycle, which would prevent the fold
+        // consumer (appended last) from ever being reached — the AofFold reply
+        // is never sent, the per-shard writer blocks on recv_blocking() forever,
+        // and all other shards' writers wait on await_outcome() indefinitely
+        // (permanent deadlock until SIGKILL). Putting the fold consumer first
+        // costs ≤ 2 extra try_pop calls per cycle (ring capacity is 4) and
+        // eliminates the starvation window.
+        if let Some(ref mut fold_cons_vec) = fold_consumers {
+            let fold_cons = std::mem::replace(&mut fold_cons_vec[id], {
+                use ringbuf::traits::Split;
+                ringbuf::HeapRb::new(1).split().1
+            });
+            consumers.insert(0, fold_cons);
+        }
         let conn_rx = mesh.take_conn_rx(id);
         let shard_cancel = cancel_token.clone();
         let shard_aof_pool = aof_pool.clone();
@@ -1299,6 +1359,8 @@ fn main() -> anyhow::Result<()> {
         let shard_pubsub_registries = all_pubsub_registries.clone();
         let shard_remote_sub_maps = all_remote_sub_maps.clone();
         let shard_affinity = affinity_tracker.clone();
+        // C1: hand the shard its ShardSlice initializer — consumed by init_shard inside run().
+        let shard_slice_init = slice_inits.remove(0);
 
         let handle = std::thread::Builder::new()
             .name(format!("shard-{}", id))
@@ -1340,6 +1402,7 @@ fn main() -> anyhow::Result<()> {
                             shard_pubsub_registries,
                             shard_remote_sub_maps,
                             shard_affinity,
+                            shard_slice_init,
                         )
                         .await;
                 });

src/persistence/aof/mod.rs

@@ -181,14 +181,25 @@ pub enum AofMessage {
     RewriteSharded(Arc<crate::shard::shared_databases::ShardDatabases>),
     /// [F6] Trigger a per-shard AOF rewrite (compaction) in the PerShard
     /// layout. Sent to EVERY per-shard writer at once. Each writer folds its
-    /// own shard (drain → lock → snapshot → write new base+incr at the
-    /// coordinator's `new_seq` → reopen), then decrements the shared
+    /// own shard (drain → AofFold SPSC → snapshot → write new base+incr at
+    /// the coordinator's `new_seq` → reopen), then decrements the shared
     /// `PerShardRewriteCoord`; the last writer commits the manifest once
     /// (single seq flip) and prunes the old generation. The synchronized seq
     /// + single commit are what make multi-shard BGREWRITEAOF crash-safe.
+    ///
+    /// `fold_producer` / `fold_notifier` are per-shard SPSC handles used for
+    /// the C4 cooperative snapshot (ShardMessage::AofFold).  Each writer
+    /// receives the producer/notifier for ITS own shard only.
     RewritePerShard {
         shard_dbs: Arc<crate::shard::shared_databases::ShardDatabases>,
         coord: Arc<PerShardRewriteCoord>,
+        /// SPSC producer into this shard's event-loop ring buffer.
+        /// Wrapped in `Mutex` so the AOF writer thread (not the shard thread)
+        /// can push the AofFold message safely.
+        fold_producer:
+            Arc<parking_lot::Mutex<ringbuf::HeapProd<crate::shard::dispatch::ShardMessage>>>,
+        /// Notifier that wakes the shard event loop after an SPSC push.
+        fold_notifier: Arc<crate::runtime::channel::Notify>,
     },
     /// Shut down the AOF writer task gracefully.
     Shutdown,