fix: bound fork-recovery replay to the nearest snapshot, never from genesis

reconcile_state_after_rollback fell back to a full replay from height 0 under
the global state lock on every rollback, freezing the node for minutes and
turning a transient one-block fork into a permanent network halt. Root: commit
337ca43 switched the boundary snapshot writer to full_snap_, but the recovery
reader (load_state_snapshot_by_height) only read the legacy state_snap_ key, so
no snapshot was ever found.

- add Storage::decode_snapshot_accounts: format-aware decoder that reads the
  canonical full_snap_ (Format A, raw accounts-CF dump) and the legacy
  state_snap_ (Format B, bincode Vec) into the account list
- reconcile now decodes the snapshot bytes already fetched by
  find_snapshot_at_or_before (no second state_snap_-only read), restoring from
  the freshest snapshot <= target so replay is bounded to <= the snapshot
  interval instead of a genesis replay
- refuse a from-0 replay above the genesis window: no usable snapshot and
  target > interval returns Err so the caller re-syncs from the canonical
  2f+1 chain instead of freezing

A transient fork above finality is now bounded and self-heals via re-sync.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: AIQnetLab

development/qnet-integration/src/node.rs

@@ -3500,46 +3500,33 @@ impl BlockchainNode {
         // Decoding is purely a CPU transformation of bytes the caller
         // already owns; doing it BEFORE we acquire the write lock keeps
         // the apply pipeline blocked for the minimum possible window.
+        // Decode the snapshot bytes already fetched by find_snapshot_at_or_before (canonical
+        // full_snap_ or legacy state_snap_). Restoring from the freshest snapshot ≤ target
+        // bounds replay to ≤ SNAPSHOT_INCREMENTAL_INTERVAL instead of a full genesis replay.
         let restored_baseline: Option<(u64, Vec<(String, qnet_state::Account)>)> =
             match snap_choice {
-                Some((snap_height, _snap_data)) => {
-                    match storage.load_state_snapshot_by_height(snap_height).await {
-                        Ok(Some((_state_root, accounts_bytes))) => {
-                            let deser = bincode::DefaultOptions::new()
-                                .with_fixint_encoding()
-                                .allow_trailing_bytes()
-                                .deserialize::<Vec<(String, qnet_state::Account)>>(&accounts_bytes)
-                                .or_else(|_| bincode::deserialize::<Vec<(String, qnet_state::Account)>>(&accounts_bytes));
-                            match deser {
-                                Ok(accounts) => Some((snap_height, accounts)),
-                                Err(e) => {
-                                    println!(
-                                        "[WARN][STATE] reconcile_deserialize_failed snap_h={} err={} action=full_replay",
-                                        snap_height, e,
-                                    );
-                                    None
-                                }
-                            }
-                        }
-                        Ok(None) => None,
+                Some((snap_height, snap_data)) => {
+                    match storage.decode_snapshot_accounts(&snap_data) {
+                        Ok(accounts) => Some((snap_height, accounts)),
                         Err(e) => {
-                            println!(
-                                "[WARN][STATE] reconcile_load_snapshot_failed snap_h={} err={:?} action=full_replay",
-                                snap_height, e,
-                            );
+                            println!("[WARN][STATE] reconcile_decode_failed snap_h={} err={:?}", snap_height, e);
                             None
                         }
                     }
                 }
-                None => {
-                    println!(
-                        "[INFO][STATE] reconcile_no_snapshot_available target={} action=full_replay",
-                        target_height,
-                    );
-                    None
-                }
+                None => None,
             };
 
+        // No usable snapshot above the genesis window ⇒ refuse a from-0 replay (it freezes the
+        // node under the state lock and resets the per-mb baseline). Return Err so the caller
+        // re-syncs from the canonical 2f+1 chain instead. A snapshot exists at every interval.
+        const SNAPSHOT_INCREMENTAL_INTERVAL: u64 = 3_600;
+        if restored_baseline.is_none() && target_height > SNAPSHOT_INCREMENTAL_INTERVAL {
+            return Err(format!(
+                "reconcile_no_usable_snapshot target={} action=resync_required", target_height
+            ));
+        }
+
         // Step 3: pre-load every microblock that needs to be replayed.
         // Doing this BEFORE we acquire the state write lock means the
         // apply pipeline only blocks during pure in-memory CPU work,

development/qnet-integration/src/storage.rs

@@ -7852,6 +7852,61 @@ impl Storage {
         }
     }
 
+    /// Decode a snapshot blob into its account list for in-memory state rebuild during
+    /// fork-recovery. Reads BOTH the canonical full_snap_ (Format A: raw accounts-CF dump)
+    /// and the legacy state_snap_ (Format B: bincode Vec). Accounts only — other CF sections
+    /// ignored. Pure (no DB). Inverse of create_state_snapshot/save_state_snapshot writers.
+    pub fn decode_snapshot_accounts(&self, snap_data: &[u8]) -> IntegrationResult<Vec<(String, qnet_state::Account)>> {
+        if snap_data.len() < 41 {
+            return Err(IntegrationError::StorageError(format!("snapshot too short: {} bytes", snap_data.len())));
+        }
+        let stored_hash = &snap_data[..32];
+        let compressed = &snap_data[40..];
+        use sha3::{Sha3_256, Digest};
+        let mut hasher = Sha3_256::new();
+        hasher.update(compressed);
+        if stored_hash != hasher.finalize().as_slice() {
+            return Err(IntegrationError::StorageError("snapshot integrity check failed".to_string()));
+        }
+        let buf = zstd::decode_all(compressed)
+            .map_err(|e| IntegrationError::StorageError(format!("snapshot decompress failed: {}", e)))?;
+        if buf.first().copied() != Some(0x02) || buf.len() < 5 {
+            return Err(IntegrationError::StorageError("snapshot wrong/short type".to_string()));
+        }
+        // probe u32 after type byte: >=10_000 ⇒ Format B (state_root bytes); else Format A version
+        let probe = u32::from_le_bytes(buf[1..5].try_into().unwrap());
+        let mut out: Vec<(String, qnet_state::Account)> = Vec::new();
+        if probe >= 10_000 {
+            // Format B: [0x02 | state_root(32) | total_supply(8) | height(8) | bincode(Vec<(addr,Account)>)]
+            let body = 1 + 32 + 8 + 8;
+            if buf.len() < body { return Err(IntegrationError::StorageError("format_b truncated".to_string())); }
+            out = bincode::deserialize(&buf[body..])
+                .map_err(|e| IntegrationError::SerializationError(format!("format_b decode: {}", e)))?;
+        } else {
+            // Format A: [0x02 | version(4) | height(8) | ts(8) | (klen|k|vlen|v)* | REWARDS_V1 ...]
+            let mut cursor = 1 + 4 + 8 + 8;
+            while cursor < buf.len() {
+                if cursor + 10 <= buf.len() && &buf[cursor..cursor + 10] == b"REWARDS_V1" { break; }
+                if cursor + 4 > buf.len() { break; }
+                let klen = u32::from_le_bytes(buf[cursor..cursor + 4].try_into().unwrap()) as usize;
+                cursor += 4;
+                if cursor + klen > buf.len() { break; }
+                let key = &buf[cursor..cursor + klen]; cursor += klen;
+                if cursor + 4 > buf.len() { break; }
+                let vlen = u32::from_le_bytes(buf[cursor..cursor + 4].try_into().unwrap()) as usize;
+                cursor += 4;
+                if cursor + vlen > buf.len() { break; }
+                let val = &buf[cursor..cursor + vlen]; cursor += vlen;
+                let addr = String::from_utf8(key.to_vec())
+                    .map_err(|e| IntegrationError::StorageError(format!("addr utf8: {}", e)))?;
+                let account = bincode::deserialize::<qnet_state::Account>(val)
+                    .map_err(|e| IntegrationError::SerializationError(format!("account decode: {}", e)))?;
+                out.push((addr, account));
+            }
+        }
+        Ok(out)
+    }
+
     pub async fn load_state_snapshot_by_height(&self, height: u64) -> IntegrationResult<Option<([u8; 32], Vec<u8>)>> {
         let snapshots_cf = self.persistent.db.cf_handle("snapshots")
             .ok_or_else(|| IntegrationError::StorageError("snapshots column family not found".to_string()))?;