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).