AIQnetLab
diff --git a/‎core/qnet-state/src/state.rs‎
Lines changed: 106 additions & 7 deletions b/‎core/qnet-state/src/state.rs‎
Lines changed: 106 additions & 7 deletions
diff --git a/‎development/qnet-integration/src/block_pipeline.rs‎
Lines changed: 51 additions & 31 deletions b/‎development/qnet-integration/src/block_pipeline.rs‎
Lines changed: 51 additions & 31 deletions
@@ -786,6 +786,14 @@ pub trait AccountStore: Send + Sync {
     /// the caller treats both identically (cache miss falls through to
     /// the canonical "account not found" path).
     fn load_account(&self, address: &str) -> Option<Account>;
+
+    /// Durable batch write of accounts. Called by the eviction sweep BEFORE dropping entries from the
+    /// cache; returns true IFF the write durably succeeded. The evictor removes ONLY a successfully-
+    /// persisted batch, so a failed persist (I/O error, or no store) keeps the accounts resident —
+    /// never silently dropping a cold mutation not yet write-through-persisted (genesis /
+    /// producer-inline), which would diverge the persistent mirror from the committed tree. Default
+    /// false (a read-only store never persists ⇒ its accounts are never evicted).
+    fn persist_accounts(&self, _accounts: &[(String, Account)]) -> bool { false }
 }
 
 pub struct StateManager {
@@ -1052,14 +1060,38 @@ impl StateManager {
             .collect();
         sorted.sort_by_key(|(_, ts)| *ts);
 
+        // Persist-before-evict: snapshot the victims' current values and flush
+        // them to the durable store BEFORE dropping from the cache, so a cold
+        // mutation not yet write-through-persisted (genesis / producer-inline)
+        // is never lost — the persistent store stays a complete mirror of the
+        // committed tree. Persist-then-remove (not the reverse) leaves no window
+        // where a concurrent read sees the address as absent.
+        let victims: Vec<(String, Account)> = sorted.iter().take(target_evict)
+            .filter_map(|(addr, _)| self.accounts.get(addr).map(|e| (addr.clone(), e.value().clone())))
+            .collect();
+        // Evict a batch ONLY when it is safe to drop from RAM: if a durable store is wired, evict iff the
+        // persist succeeded (a failed write keeps the victims resident so a cold mutation is never lost
+        // and the persistent mirror never diverges from the committed tree — the next sweep retries). If
+        // NO store is wired (cache-only mode: tests/tooling) there is no durable mirror to diverge from,
+        // so eviction is the intended cache bound. Production always wires the store before raising cap.
+        let persisted = if victims.is_empty() {
+            true
+        } else {
+            match *self.disk_store.read() {
+                Some(ref store) => store.persist_accounts(&victims),
+                None => true,
+            }
+        };
         let mut evicted = 0usize;
-        for (addr, _) in sorted.iter().take(target_evict) {
-            self.accounts.remove(addr);
-            self.last_access.remove(addr);
-            evicted = evicted.saturating_add(1);
-        }
-        if evicted > 0 {
-            self.evictions_total.fetch_add(evicted as u64, std::sync::atomic::Ordering::Relaxed);
+        if persisted {
+            for (addr, _) in &victims {
+                self.accounts.remove(addr);
+                self.last_access.remove(addr);
+                evicted = evicted.saturating_add(1);
+            }
+            if evicted > 0 {
+                self.evictions_total.fetch_add(evicted as u64, std::sync::atomic::Ordering::Relaxed);
+            }
         }
         evicted
     }
@@ -2244,6 +2276,13 @@ mod cache_tests {
             self.load_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
             self.data.read().get(address).cloned()
         }
+        fn persist_accounts(&self, accounts: &[(String, Account)]) -> bool {
+            let mut g = self.data.write();
+            for (addr, acct) in accounts {
+                g.insert(addr.clone(), acct.clone());
+            }
+            true
+        }
     }
 
     fn make_account(balance: u64) -> Account {
@@ -2306,6 +2345,66 @@ mod cache_tests {
         assert_eq!(store.loads(), 1);
     }
 
+    /// Persist-before-evict: a cold mutation that never went through the
+    /// write-through path (genesis / producer-inline) must be flushed to the
+    /// durable store at eviction, so durable ∪ cache loses nothing — the
+    /// snapshot-completeness invariant at scale (past the LRU cap).
+    #[test]
+    fn test_persist_before_evict_no_data_loss() {
+        let sm = StateManager::new();
+        let store = MockStore::new();
+        sm.set_disk_store(store.clone() as Arc<dyn AccountStore>);
+        sm.set_cache_capacity(10);
+
+        for i in 0..25u64 {
+            let addr = format!("acct_{:03}", i);
+            sm.accounts.insert(addr.clone(), make_account(i));
+            sm.last_access.insert(addr, i); // ascending ⇒ lowest i evicted first
+        }
+        assert_eq!(sm.accounts.len(), 25);
+
+        let evicted = sm.evict_cold_accounts();
+        assert_eq!(evicted, 15, "must evict down to the cap");
+        assert_eq!(sm.accounts.len(), 10);
+
+        // No account lost: each is in the cache OR persisted on disk.
+        for i in 0..25u64 {
+            let addr = format!("acct_{:03}", i);
+            let in_cache = sm.accounts.contains_key(&addr);
+            let on_disk = store.data.read().contains_key(&addr);
+            assert!(in_cache || on_disk, "account {} lost on eviction", addr);
+        }
+        // The 15 oldest were evicted ⇒ must have been persisted before removal.
+        for i in 0..15u64 {
+            let addr = format!("acct_{:03}", i);
+            assert!(store.data.read().contains_key(&addr), "evicted {} must be persisted", addr);
+        }
+    }
+
+    /// A wired durable store whose write FAILS must NOT drop accounts from RAM — else the persistent
+    /// mirror diverges from the committed tree (snapshot state_root mismatch). Keep them resident.
+    struct FailingStore;
+    impl AccountStore for FailingStore {
+        fn load_account(&self, _address: &str) -> Option<Account> { None }
+        fn persist_accounts(&self, _accounts: &[(String, Account)]) -> bool { false } // always fails
+    }
+
+    #[test]
+    fn test_evict_keeps_accounts_when_persist_fails() {
+        let sm = StateManager::new();
+        sm.set_disk_store(Arc::new(FailingStore) as Arc<dyn AccountStore>);
+        sm.set_cache_capacity(10);
+        for i in 0..25u64 {
+            let addr = format!("acct_{:03}", i);
+            sm.accounts.insert(addr.clone(), make_account(i));
+            sm.last_access.insert(addr, i);
+        }
+        assert_eq!(sm.accounts.len(), 25);
+        let evicted = sm.evict_cold_accounts();
+        assert_eq!(evicted, 0, "must not evict when the durable persist fails");
+        assert_eq!(sm.accounts.len(), 25, "all accounts stay resident on persist failure");
+    }
+
     #[test]
     fn test_warm_account_genuine_miss_returns_false() {
         let sm = StateManager::new();
 
@@ -2391,27 +2391,34 @@ impl BlockPipeline {
             // path; if hung, watchdog will surface it.
             metrics.mark_apply_op(height, PIPELINE_OP_APPLY_DEDUP);
             let dedup_start = std::time::Instant::now();
-            // v15.6: Dedup check runs on the blocking pool. The RocksDB lookup
-            // on a hot row competes with the same column family the apply
-            // stage writes to a few microseconds later; running it on the
-            // async path made one slow read freeze the entire stage. The
-            // tokio::task::spawn_blocking handoff is cheap (single channel
-            // hop) and isolates this I/O from runtime liveness.
-            let storage_for_dedup = ctx.storage.clone();
-            let already_applied = match tokio::task::spawn_blocking(move || {
-                storage_for_dedup.load_microblock(height)
-                    .map(|opt| opt.is_some())
-                    .unwrap_or(false)
-            }).await {
-                Ok(v) => v,
-                Err(join_err) => {
-                    if is_warn() {
-                        println!(
-                            "[WARN][PIPELINE] apply_dedup_join_err h={} err={}",
-                            height, join_err
-                        );
+            // Apply is strictly sequential by height and publishes the applied frontier in
+            // LOCAL_BLOCKCHAIN_HEIGHT at commit. A block strictly ABOVE that frontier cannot be a
+            // duplicate, so the common path is answered with an O(1) atomic read — NO hot-path
+            // RocksDB lookup (a storage read here contends with the same CF the apply stage writes
+            // microseconds later, and one slow read under a maintenance-flush/compaction storm
+            // froze the whole stage). Only a re-delivery (height <= frontier) consults storage, off
+            // the hot path on the blocking pool.
+            let applied_tip = crate::unified_p2p::LOCAL_BLOCKCHAIN_HEIGHT
+                .load(std::sync::atomic::Ordering::Acquire);
+            let already_applied = if height > applied_tip {
+                false
+            } else {
+                let storage_for_dedup = ctx.storage.clone();
+                match tokio::task::spawn_blocking(move || {
+                    storage_for_dedup.load_microblock(height)
+                        .map(|opt| opt.is_some())
+                        .unwrap_or(false)
+                }).await {
+                    Ok(v) => v,
+                    Err(join_err) => {
+                        if is_warn() {
+                            println!(
+                                "[WARN][PIPELINE] apply_dedup_join_err h={} err={}",
+                                height, join_err
+                            );
+                        }
+                        false
                     }
-                    false
                 }
             };
             let dedup_elapsed = dedup_start.elapsed();
@@ -3063,22 +3070,35 @@ impl BlockPipeline {
                 }
             }
 
-            // Canonical boundary snapshot (raw accounts-CF dump via create_state_snapshot) at every
-            // SNAPSHOT_INCREMENTAL_INTERVAL, on EVERY node's apply path (not just the producer) so a
-            // cold joiner can fast-sync from any peer. This is the SAME representation the macroblock
-            // snapshot_root binds (compute_canonical_state_root over the accounts CF) → restore
-            // reproduces the bound root. Off-reactor; WARN-only, never blocks liveness.
+            // Canonical boundary snapshot at every SNAPSHOT_INCREMENTAL_INTERVAL, on EVERY node's apply
+            // path so a cold joiner can fast-sync from any peer. Pin a frozen DB view at `height`
+            // SYNCHRONOUSLY here — the serial apply loop has not started H+1, so the flush + snapshot
+            // capture exactly state_root@H. With persist-before-evict the pinned accounts CF is the
+            // COMPLETE committed tree leaf set (hot ∪ evicted), so a cold joiner's recompute reproduces
+            // the bound root past the LRU cap. The heavy serialization runs off-reactor on the frozen view.
             const SNAPSHOT_INCREMENTAL_INTERVAL: u64 = 3_600;
             if height > 0 && height % SNAPSHOT_INCREMENTAL_INTERVAL == 0 {
-                let storage_for_snapshot = ctx.storage.clone();
-                let snapshot_height = height;
-                tokio::spawn(async move {
-                    if let Err(e) = storage_for_snapshot.create_state_snapshot(snapshot_height).await {
+                let snapshot_accounts = ctx.state.read().await.get_all_accounts();
+                match ctx.storage.prepare_snapshot_view(&snapshot_accounts) {
+                    Ok(view) => {
+                        let storage_for_snapshot = ctx.storage.clone();
+                        let snapshot_height = height;
+                        tokio::spawn(async move {
+                            if let Err(e) = storage_for_snapshot
+                                .create_state_snapshot(snapshot_height, view).await
+                            {
+                                if is_warn() {
+                                    println!("[WARN][PIPELINE] snapshot_create_failed h={} err={:?}", snapshot_height, e);
+                                }
+                            }
+                        });
+                    }
+                    Err(e) => {
                         if is_warn() {
-                            println!("[WARN][PIPELINE] snapshot_create_failed h={} err={:?}", snapshot_height, e);
+                            println!("[WARN][PIPELINE] snapshot_prepare_failed h={} err={:?}", height, e);
                         }
                     }
-                });
+                }
             }
 
             // ────────────────────────────────────────────────────────────────