fix: fail-closed fork recovery + apply circuit-breaker + O(N) committee + cold-join snapshot-first

- recovery (node.rs): reconcile fail-closed — accept only when the recomputed state root
  equals the 2f+1 macroblock snapshot_root; pre-finality / missing-anchor / no-binding /
  mismatch all resync. On unproven reconcile: discard and clean QC-verified state-sync,
  never replay on a contaminated base.
- pipeline (block_pipeline.rs): apply circuit-breaker — escalate to fork recovery after 3
  consecutive state_root_mismatch (across heights) instead of looping on a bad base.
- committee (node.rs): O(N) select_nth_unstable_by quickselect — identical deterministic
  set, removes the 100k-eligible full-sort ceiling.
- cold-join (sync_manager.rs): snapshot-first in execute_sync; desync detection + sync
  target floored to the QC-verified frontier, not stale per-peer height.
- activation (storage.rs): wallet_is_genesis_node — exempt genesis self-activation (no burn)
  from the NodeActivation burn-gate.

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

Author: AIQnetLab

development/qnet-integration/src/block_pipeline.rs

@@ -83,6 +83,26 @@ pub fn take_fork_recovery_signal() -> Option<u64> {
     }
 }
 
+// Apply-stage circuit-breaker: consecutive apply failures (state_root_mismatch) with no
+// clean apply in between. Repeated failure means the local base is contaminated; the node
+// stops re-applying onto it (the wedge) and escalates to fail-closed fork recovery. Counted
+// across heights so a mismatch that hops to the next height cannot reset the count and dodge
+// the breaker. Cleared on any successful apply.
+static APPLY_MISMATCH_COUNT: AtomicU64 = AtomicU64::new(0);
+const APPLY_MISMATCH_BREAKER: u64 = 3;
+
+/// Record an apply failure; returns true once it trips the breaker.
+fn record_apply_mismatch() -> bool {
+    APPLY_MISMATCH_COUNT.fetch_add(1, Ordering::Relaxed) + 1 >= APPLY_MISMATCH_BREAKER
+}
+
+/// Reset the breaker after a clean apply.
+fn clear_apply_mismatch() {
+    if APPLY_MISMATCH_COUNT.load(Ordering::Relaxed) != 0 {
+        APPLY_MISMATCH_COUNT.store(0, Ordering::Relaxed);
+    }
+}
+
 // Distinct-peer witness tracker for microblock minority-fork detection.
 // Height → set of distinct peer_ids that reported hash_chain_break there.
 // DETECTION threshold is f+1, NOT 2f+1: a node on a minority fork cannot
@@ -2239,6 +2259,7 @@ impl BlockPipeline {
                         matches!(t.tx_type, qnet_state::TransactionType::NodeActivation { .. })
                             && !this_block_burned.contains(&t.from)
                             && !storage.wallet_is_burn_registered(&t.from)
+                            && !storage.wallet_is_genesis_node(&t.from) // genesis self-activates w/o burn
                     });
                     if unbacked {
                         if is_warn() {
@@ -2542,11 +2563,22 @@ impl BlockPipeline {
                     // (e.g. a contaminated/orphaned base), not the peer's fault. Striking honest
                     // peers poisoned the pool and blocked cold-start recovery. Genuine forks are
                     // resolved by fork-choice; malice by on-chain analyze_chain_for_slashing.
+
+                    // Circuit-breaker: re-applying the same canonical block onto a contaminated
+                    // base mismatches forever (the wedge). On threshold, escalate to fork
+                    // recovery — which is fail-closed and ends in a clean QC-verified state-sync.
+                    if record_apply_mismatch() {
+                        FORK_RECOVERY_HEIGHT.store(height.saturating_sub(1).max(1), Ordering::SeqCst);
+                        if is_warn() {
+                            println!("[WARN][PIPELINE] apply_breaker_tripped h={} action=fork_recovery", height);
+                        }
+                    }
                     metrics.mark_apply_idle();
                     continue;
                 }
 
                 // v14.8: Successful apply — clear any past strikes for this peer.
+                clear_apply_mismatch();
                 if let Some(ref p2p) = ctx.unified_p2p {
                     p2p.record_apply_success(&block.from_peer);
                 }

development/qnet-integration/src/node.rs

@@ -3676,31 +3676,43 @@ impl BlockchainNode {
         // reject so the caller re-syncs from canonical instead of forking.
         // No binding at this boundary ⇒ cannot verify, accept with warning
         // (legitimate at pre-binding/early heights).
+        // Fail-closed by default: the ONLY accepted outcome is a recomputed state_root that
+        // equals the 2f+1-bound macroblock snapshot_root (Pattern C). Every other path —
+        // pre-finality height, missing/undecodable anchor, no binding, mismatch, recompute
+        // error — returns Err so the caller discards and resyncs from canonical QC state.
         let mb_idx = target_height / 90;
-        if mb_idx > 0 {
-            if let Ok(Some(mb_bytes)) = storage.get_macroblock_by_height(mb_idx) {
-                if let Ok(mb) = bincode::deserialize::<qnet_state::MacroBlock>(&mb_bytes) {
-                    if let Some(expected_root) = mb.consensus_data.snapshot_root {
-                        match storage.compute_canonical_state_root(target_height) {
-                            Ok(computed) if computed == expected_root => {
-                                println!("[INFO][STATE] reconcile_verified mb={} target={} root={} pattern=C",
-                                         mb_idx, target_height, hex::encode(&computed[..8]));
-                            }
-                            Ok(computed) => {
-                                return Err(format!(
-                                    "reconcile_root_mismatch target={} mb={} expected={} computed={} action=resync",
-                                    target_height, mb_idx,
-                                    hex::encode(&expected_root[..8]), hex::encode(&computed[..8]),
-                                ));
-                            }
-                            Err(e) => {
-                                println!("[WARN][STATE] reconcile_verify_compute_err target={} err={:?}", target_height, e);
-                            }
-                        }
-                    } else {
-                        println!("[WARN][STATE] reconcile_unverified mb={} reason=no_snapshot_root_binding", mb_idx);
-                    }
-                }
+        if mb_idx == 0 {
+            // h<90: no finalized macroblock to prove canonicity ⇒ resync (block-sync from
+            // genesis is cheap pre-finality). Never accept unverified recovery state.
+            return Err(format!("reconcile_pre_finality target={} action=resync", target_height));
+        }
+        let mb_bytes = match storage.get_macroblock_by_height(mb_idx) {
+            Ok(Some(b)) => b,
+            _ => return Err(format!("reconcile_anchor_unavailable mb={} target={} action=resync", mb_idx, target_height)),
+        };
+        let mb = match bincode::deserialize::<qnet_state::MacroBlock>(&mb_bytes) {
+            Ok(m) => m,
+            Err(e) => return Err(format!("reconcile_anchor_decode mb={} err={:?} action=resync", mb_idx, e)),
+        };
+        let expected_root = match mb.consensus_data.snapshot_root {
+            Some(r) => r,
+            None => return Err(format!("reconcile_no_binding mb={} target={} action=resync", mb_idx, target_height)),
+        };
+        match storage.compute_canonical_state_root(target_height) {
+            Ok(computed) if computed == expected_root => {
+                println!("[INFO][STATE] reconcile_verified mb={} target={} root={} pattern=C",
+                         mb_idx, target_height, hex::encode(&computed[..8]));
+            }
+            Ok(computed) => {
+                return Err(format!(
+                    "reconcile_root_mismatch target={} mb={} expected={} computed={} action=resync",
+                    target_height, mb_idx, hex::encode(&expected_root[..8]), hex::encode(&computed[..8]),
+                ));
+            }
+            Err(e) => {
+                return Err(format!(
+                    "reconcile_verify_unavailable target={} mb={} err={:?} action=resync", target_height, mb_idx, e,
+                ));
             }
         }
         Ok(())
@@ -4219,7 +4231,11 @@ impl BlockchainNode {
             // activation cost. Mirrors select_consensus_committee (same beacon-seeded VRF, one
             // tier up). node_id is a total-order tiebreak for the (cryptographically
             // unreachable) SHA3-collision case.
-            eligible.sort_by(|a, b| {
+            // O(N) partial-select (quickselect) instead of a full O(N log N) sort of the whole
+            // pool — the committee-selection ceiling at 100k+ eligible. Yields the identical set
+            // (the MAX_VALIDATORS lowest VRF scores; node_id total-order tiebreak), then sort only
+            // the selected for deterministic order. Reached only when len > MAX_VALIDATORS.
+            eligible.select_nth_unstable_by(MAX_VALIDATORS, |a, b| {
                 vrf_scores[&a.node_id].cmp(&vrf_scores[&b.node_id])
                     .then_with(|| a.node_id.cmp(&b.node_id))
             });
@@ -13914,10 +13930,20 @@ impl BlockchainNode {
                                             &storage,
                                             rollback_to,
                                         ).await {
+                                            // Reconcile could not PROVE the rebuilt state canonical.
+                                            // Never proceed on unverified state: discard it and reload a
+                                            // 2f+1-QC-bound snapshot (fast_sync verifies the binding,
+                                            // fail-closed), then the tail re-syncs verify-then-apply.
                                             println!(
-                                                "[ERR][STATE] reconcile_after_pipeline_fork_failed target={} err={} action=resync_required",
+                                                "[WARN][STATE] reconcile_unproven target={} err={} action=clean_state_sync",
                                                 rollback_to, e,
                                             );
+                                            let tip = crate::node::qc_verified_frontier_height()
+                                                .max(p2p.get_best_peer_height());
+                                            match storage.fast_sync_with_snapshot(p2p, tip).await {
+                                                Ok(()) => println!("[INFO][STATE] clean_state_sync_ok target={}", tip),
+                                                Err(se) => println!("[WARN][STATE] clean_state_sync_failed err={:?} fallback=block_sync", se),
+                                            }
                                         } else {
                                             println!(
                                                 "[INFO][STATE] reconcile_after_pipeline_fork_ok target={}",
@@ -16505,7 +16531,11 @@ impl BlockchainNode {
                         }).collect();
                         txs.retain(|t| {
                             if matches!(t.tx_type, qnet_state::TransactionType::NodeActivation { .. }) {
-                                let backed = this_block_burned.contains(&t.from) || storage.wallet_is_burn_registered(&t.from);
+                                // Genesis nodes self-activate without a 1DEV burn (they ARE the bootstrap),
+                                // so exempt them — same genesis exemption the registration burn-gate uses.
+                                let backed = this_block_burned.contains(&t.from)
+                                    || storage.wallet_is_burn_registered(&t.from)
+                                    || storage.wallet_is_genesis_node(&t.from);
                                 if !backed && is_warn() { println!("[WARN][MB] drop_unbacked_activation h={}", next_block_height); }
                                 backed
                             } else { true }

development/qnet-integration/src/storage.rs

@@ -6117,6 +6117,21 @@ impl Storage {
         }
     }
 
+    /// True iff `wallet` belongs to a GENESIS bootstrap node — its registration maps (wallet_ reverse
+    /// index) to a node_id the genesis constants recognise. Genesis nodes are protocol-minted and
+    /// activate WITHOUT a 1DEV burn (they ARE the bootstrap), so the NodeActivation burn-gate must
+    /// exempt them — mirroring exactly the registration burn-attestation gate's genesis exemption
+    /// (is_legacy_genesis_node). Without this, a genesis self-activation (empty burn) is wrongly dropped.
+    pub fn wallet_is_genesis_node(&self, wallet: &str) -> bool {
+        let cf = match self.persistent.db.cf_handle("node_registry") { Some(c) => c, None => return false };
+        match self.persistent.db.get_cf(&cf, format!("wallet_{}", wallet).as_bytes()) {
+            Ok(Some(v)) => serde_json::from_slice::<serde_json::Value>(&v).ok()
+                .and_then(|j| j["node_id"].as_str().map(|s| crate::genesis_constants::is_legacy_genesis_node(s)))
+                .unwrap_or(false),
+            _ => false,
+        }
+    }
+
     /// Rebuild the committed burn→wallet index (cbw_) DETERMINISTICALLY from the chain-confirmed
     /// node_ registry entries, considering ONLY registrations with reg_height <= up_to_height.
     /// cbw is a pure DERIVED index, never deleted per-block — so a snapshot/fast-sync join (restores
@@ -10811,6 +10826,11 @@ mod v32_9_pattern_c_tests {
         assert!(!storage.wallet_is_burn_registered("walletG"), "empty-burn registration ⇒ not a burn proof");
         // A raw activation from an unregistered wallet is rejected.
         assert!(!storage.wallet_is_burn_registered("walletX"), "no registration ⇒ activation rejected");
+        // Genesis exemption: genesis self-activates without a 1DEV burn, so the gate must let it through
+        // via wallet_is_genesis_node (mirrors the registration burn-gate's is_legacy_genesis_node).
+        assert!(storage.wallet_is_genesis_node("walletG"), "genesis wallet ⇒ activation exempt");
+        assert!(!storage.wallet_is_genesis_node("walletA"), "non-genesis super ⇒ NOT genesis-exempt");
+        assert!(!storage.wallet_is_genesis_node("walletX"), "unregistered ⇒ NOT genesis-exempt");
     }
 
     #[test]

development/qnet-integration/src/sync_manager.rs

@@ -345,7 +345,12 @@ impl SyncManager {
     /// Check if we're behind and need to sync.
     async fn check_desync(&self) {
         let snap = self.coordinator.snapshot();
-        let network_h = self.p2p.get_best_peer_height();
+        // D1: drive the desync decision off the QC-verified frontier (bootstrap-cross-checked),
+        // NOT the raw BEST_PEER_HEIGHT atomic. That scalar is monotonic (never decays) and is
+        // polluted by stale/frozen per-peer heights, so a phantom value could spuriously trigger
+        // or a stale-low one suppress sync. detect_network_height floors to the frontier — the same
+        // source the SyncToNetwork command path uses — so stale peers can no longer mis-drive it.
+        let network_h = self.detect_network_height().await;
         let local_h = snap.chain_height;
 
         if network_h > local_h + self.config.auto_sync_gap {
@@ -391,7 +396,18 @@ impl SyncManager {
     ///   would verify (missing previous_hash for genesis) — triggering a cycle
     ///   of deferred_full drops until genesis eventually arrives randomly.
     async fn execute_sync(&self, target: u64) {
-        let local_h = self.coordinator.chain_height();
+        let mut local_h = self.coordinator.chain_height();
+
+        // D1: never let an unverified scalar drive the bulk target. Floor it to the QC-verified
+        // finality frontier (authoritative); an unverified hint may only add the ≤2-macroblock
+        // unsealed tail above it. frontier==0 (fresh genesis, h<90) ⇒ target as-is so the
+        // 5-genesis bootstrap is never blocked. Mirrors detect_network_height; protects every
+        // caller (SyncTo / SyncToNetwork / check_desync / snapshot fast-path) uniformly.
+        let target = {
+            let frontier = crate::node::qc_verified_frontier_height();
+            if frontier == 0 { target }
+            else { std::cmp::max(frontier, std::cmp::min(target, frontier.saturating_add(180))) }
+        };
 
         if local_h >= target {
             if is_debug() {
@@ -425,6 +441,48 @@ impl SyncManager {
                      local_h, target, target - local_h, peer_count);
         }
 
+        // ─────────────────────────────────────────────────────────────────────
+        // COLD-JOIN SNAPSHOT FAST-PATH (THE ROOT FIX). A fresh / far-behind node
+        // CANNOT converge block-by-block — the pipelined loop below only closes a
+        // SMALL gap; for a large one, block production outpaces the joiner and the
+        // gap grows without bound (a 6k-block joiner never catches up). The node MUST
+        // restore a remote state snapshot FIRST (jump to ~tip), then the loop syncs
+        // only the residual tail. This step existed in the legacy sync path but was
+        // LOST when SyncManager replaced it (the old fast_sync_with_snapshot call
+        // sites became dead/unreachable), so every real super-node fell to block-by-
+        // block and never onboarded. The snapshot DOWNLOAD + 2f+1-QC binding +
+        // microblock-hash backfill all already exist in storage and are correct
+        // (verify_snapshot_consensus_binding is QC-anchored, fail-close) — they were
+        // simply never INVOKED on the live cold-start engine. Fire on a cold join
+        // (local==0) or a large gap; on failure (no network snapshot yet — e.g. a
+        // sub-interval fresh genesis) fall through to the block-by-block path below.
+        // On success local_h is advanced so the genesis-h=0 fetch is skipped and the
+        // loop only fills the tail. Same proven call the legacy node.rs path used.
+        const SNAPSHOT_FAST_PATH_GAP: u64 = 1_500;
+        if local_h == 0 || target.saturating_sub(local_h) > SNAPSHOT_FAST_PATH_GAP {
+            match self.storage.fast_sync_with_snapshot(&self.p2p, target).await {
+                Ok(()) => {
+                    let restored = self.storage.get_chain_height().unwrap_or(local_h);
+                    if restored > local_h {
+                        local_h = restored;
+                        self.progress_height.store(restored, Ordering::Relaxed);
+                        crate::unified_p2p::LOCAL_BLOCKCHAIN_HEIGHT.store(restored, Ordering::Release);
+                        if is_info() {
+                            println!("[INFO][SYNC] snapshot_restored h={} target={} tail={}",
+                                     restored, target, target.saturating_sub(restored));
+                        }
+                    } else if is_info() {
+                        println!("[INFO][SYNC] snapshot_no_advance local={} — fallback block_sync", local_h);
+                    }
+                }
+                Err(e) => {
+                    if is_info() {
+                        println!("[INFO][SYNC] snapshot_unavailable reason={:?} fallback=block_sync", e);
+                    }
+                }
+            }
+        }
+
         // Adaptive-window + credit-based backpressure config. HONEST NOTE:
         // initial choices, not yet measured under load — safe bounds (never
         // exceed pipeline capacity) but may leave throughput on the table;