AIQnetLab
diff --git a/‎core/qnet-consensus/src/consensus_crypto.rs‎
Lines changed: 17 additions & 0 deletions b/‎core/qnet-consensus/src/consensus_crypto.rs‎
Lines changed: 17 additions & 0 deletions
diff --git a/‎development/qnet-integration/src/block_pipeline.rs‎
Lines changed: 146 additions & 12 deletions b/‎development/qnet-integration/src/block_pipeline.rs‎
Lines changed: 146 additions & 12 deletions
diff --git a/‎development/qnet-integration/src/genesis_constants.rs‎
Lines changed: 120 additions & 0 deletions b/‎development/qnet-integration/src/genesis_constants.rs‎
Lines changed: 120 additions & 0 deletions
@@ -349,6 +349,23 @@ lazy_static::lazy_static! {
         parking_lot::RwLock::new(std::collections::HashMap::new());
 }
 
+/// Read-only access to the genesis anchor for a single identity. Returns
+/// None when no anchor map is installed (cold boot before anchor file is
+/// loaded) or when `node_id` is not a genesis identity.
+///
+/// Used by the integration layer at `initialize_wallet_identity` to refuse
+/// boot when the locally-loaded keypair does not match the anchored PK,
+/// preventing the v15.x pk_mismatch class of incidents.
+pub fn get_consensus_pk_anchor(node_id: &str) -> Option<Vec<u8>> {
+    GENESIS_ANCHOR_PKS.read().get(node_id).cloned()
+}
+
+/// Number of installed genesis anchors. 0 when no anchor file has been
+/// loaded yet — used by callers to decide whether to enforce strict binding.
+pub fn genesis_anchor_pks_len() -> usize {
+    GENESIS_ANCHOR_PKS.read().len()
+}
+
 /// Install the genesis anchor PK map. Called exactly once at process start
 /// by the integration layer, BEFORE any `register_consensus_pk_with_proof`
 /// call, with the deterministic genesis PKs for the 5 anchor nodes.
 
@@ -122,15 +122,45 @@ static HASH_CHAIN_BREAK_WITNESSES: once_cell::sync::Lazy<
 > = once_cell::sync::Lazy::new(dashmap::DashMap::new);
 
 /// Record that `peer_id` reported a hash_chain_break at `height`.
-/// If the set of distinct witnesses reaches f+1 (not 2f+1), signal fork
-/// recovery. f+1 is the "at least one honest witness" threshold and is
-/// the canonical bar for fork DETECTION (as opposed to the 2f+1 COMMIT
-/// threshold).
+///
+/// v16.1: SPLIT THRESHOLD MODEL. Industry-standard BFT chains separate
+/// fork DETECTION (advisory signal, low threshold) from fork ROLLBACK
+/// (state-mutating action, supermajority threshold):
+///
+///   * f+1 = "at least one honest witness exists" — sufficient to RAISE
+///     a soft-warning signal, prompt deeper investigation, increase peer
+///     diversity in resync sources. NEVER rolls back state.
+///
+///   * 2f+1 = "byzantine supermajority" — required for any state
+///     mutation including destructive rollback. Below this threshold
+///     a malicious f-node colluding with f-1 honest witnesses on a
+///     transient timestamp/round mismatch could force everyone else to
+///     unwind their canonical chain.
+///
+/// Forensic motivation (v15.x h=334..h=368 cascade):
+///   The legacy code used f+1 (=2 for n=5) for both detection AND
+///   destructive rollback. A single byzantine producer (node_001 with
+///   pk_mismatch broken signatures) plus one honest peer reporting a
+///   harmless `block_round=2` divergence was enough to roll the local
+///   tip back. Resync then re-downloaded the SAME forked branch from
+///   the same peer set (no canonical-chain filter), and the witness
+///   counter trip again at the new height — repeat for hours, walking
+///   the chain backwards.
+///
+///   Splitting the thresholds plus the peer-cooldown (Phase 4.B) breaks
+///   the cascade: a single dishonest peer cannot induce rollback even
+///   if f honest peers happen to agree on the same wrong-looking hash;
+///   the rollback path requires a real BFT-supermajority of distinct
+///   peer_ids.
 ///
 /// Rate-limit semantics: once FORK_RECOVERY_HEIGHT is non-zero we don't
 /// overwrite it with a different (lower) height — the main loop consumes
 /// it first. This prevents flapping when two heights both accumulate
 /// witnesses during a partition.
+///
+/// Scalability: per-height witness sets are bounded by the active
+/// validator-set count (≤ MAX_VALIDATORS = 1000 in committee). Cleanup
+/// sweep evicts entries below current chain tip.
 pub fn record_hash_chain_break_witness(height: u64, peer_id: &str) {
     if peer_id.is_empty() || peer_id == "self" {
         return;
@@ -150,14 +180,43 @@ pub fn record_hash_chain_break_witness(height: u64, peer_id: &str) {
         let n = qnet_consensus::consensus_crypto::consensus_pk_registry_len();
         if n >= 3 { n } else { 5 }
     };
-    // f+1 = ceil(n/3): guarantees at least one honest witness.
+
+    // f+1 = ceil(n/3): "at least one honest witness" — DETECTION ONLY.
     let threshold_f_plus_1 = (total_validators.saturating_add(2)) / 3;
-    // Floor at 2 so at any registry size ≥ 4 the threshold is ≥ 2; below
-    // 4 we still need at least 2 distinct reporters to avoid single-peer
-    // false positives.
-    let threshold = threshold_f_plus_1.max(2);
+    let detection_threshold = threshold_f_plus_1.max(2);
+
+    // 2f+1 = canonical BFT supermajority — destructive ROLLBACK threshold.
+    // Formula `(2n+2)/3` is the canonical ceiling form; works uniformly
+    // for n=5 as well as n=1_000_000.
+    let threshold_2f_plus_1 = ((total_validators.saturating_mul(2)).saturating_add(2)) / 3;
+    let rollback_threshold = threshold_2f_plus_1.max(3);
+
+    // Stage 1: SOFT detection signal at f+1.
+    // Logged once per crossing so operators see partial agreement without
+    // triggering rollback. The signal is also picked up by Phase 4.B
+    // peer-cooldown which tags every peer that supplied a forked block at
+    // this height as "potentially-byzantine" for canonical-aware sync
+    // peer selection — without losing those peers as gossip sources.
+    if witnesses == detection_threshold && witnesses < rollback_threshold {
+        if is_warn() {
+            println!(
+                "[WARN][PIPELINE] fork_detection_signal h={} witnesses={} threshold_f_plus_1={} action=advisory_only_no_rollback",
+                height, witnesses, detection_threshold
+            );
+        }
+        // Mark all current witnesses as "fork-source" so the canonical-
+        // aware sync peer selection prefers other peers when refilling
+        // the local chain at this height.
+        if let Some(set) = HASH_CHAIN_BREAK_WITNESSES.get(&height) {
+            for w in set.iter() {
+                mark_peer_as_fork_source(w.key());
+            }
+        }
+    }
 
-    if witnesses >= threshold {
+    // Stage 2: HARD destructive rollback at 2f+1.
+    // Only at this point do we commit to actually deleting blocks.
+    if witnesses >= rollback_threshold {
         let rollback_to = height.saturating_sub(1);
         // Only raise the signal — never lower. The main loop consumes it
         // under the same atomic swap that clears the tracker.
@@ -166,14 +225,89 @@ pub fn record_hash_chain_break_witness(height: u64, peer_id: &str) {
             FORK_RECOVERY_HEIGHT.store(rollback_to, Ordering::SeqCst);
             if is_warn() {
                 println!(
-                    "[WARN][PIPELINE] minority_fork_detected h={} rollback_to={} witnesses={} threshold={} (f+1)",
-                    height, rollback_to, witnesses, threshold
+                    "[WARN][PIPELINE] minority_fork_confirmed h={} rollback_to={} witnesses={} threshold_2f_plus_1={} action=destructive_rollback",
+                    height, rollback_to, witnesses, rollback_threshold
                 );
             }
         }
     }
 }
 
+// ═══════════════════════════════════════════════════════════════════════════
+// v16.1: FORKED PEER COOLDOWN
+// ═══════════════════════════════════════════════════════════════════════════
+// Peers that supplied blocks of a branch we just rolled back from (or which
+// triggered the f+1 fork-detection signal) are tagged here for a bounded
+// cooldown window. The canonical-aware sync peer selector reads this map
+// and de-prioritises tagged peers until the cooldown expires — letting the
+// resync pull from peers on the canonical branch instead of refetching
+// the same forked blocks.
+//
+// Bounded retention: 5-minute cooldown per peer. Auto-evicted on next
+// fork event for that peer (refresh) or via the periodic sweep below.
+// At 100k super-node deployment this map is bounded by the union of
+// recent fork participants — typically << 1000 entries.
+// ═══════════════════════════════════════════════════════════════════════════
+
+const FORKED_PEER_COOLDOWN_MS: u64 = 5 * 60 * 1000; // 5 min
+
+static FORKED_PEER_COOLDOWN: once_cell::sync::Lazy<dashmap::DashMap<String, u64>> =
+    once_cell::sync::Lazy::new(dashmap::DashMap::new);
+
+/// Mark `peer_id` as having supplied a forked-branch block. Used by the
+/// canonical-aware sync peer selector to prefer other peers during the
+/// cooldown window. Idempotent — refreshes timestamp on repeated hits.
+pub fn mark_peer_as_fork_source(peer_id: &str) {
+    if peer_id.is_empty() || peer_id == "self" {
+        return;
+    }
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+    FORKED_PEER_COOLDOWN.insert(peer_id.to_string(), now_ms);
+}
+
+/// Returns true while `peer_id` is within the fork-cooldown window. The
+/// canonical-aware sync peer selector skips peers for which this returns
+/// true; if the entire candidate set is in cooldown, the selector falls
+/// back to the full set rather than starving sync (preferring suspect
+/// peers over no peers at all when liveness is at stake).
+pub fn is_peer_in_fork_cooldown(peer_id: &str) -> bool {
+    let entry = match FORKED_PEER_COOLDOWN.get(peer_id) {
+        Some(e) => e,
+        None => return false,
+    };
+    let marked_at = *entry.value();
+    drop(entry);
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+    let in_cooldown = now_ms.saturating_sub(marked_at) < FORKED_PEER_COOLDOWN_MS;
+    if !in_cooldown {
+        // Lazy eviction — opportunistically clean expired entries on
+        // every read. Avoids a separate cleanup task at the cost of a
+        // single DashMap remove per expiration check.
+        FORKED_PEER_COOLDOWN.remove(peer_id);
+    }
+    in_cooldown
+}
+
+/// Periodic sweep called from the existing cleanup task. Removes entries
+/// older than the cooldown window so the map stays bounded under sustained
+/// fork activity. O(N) over current map size; runs at low cadence (the
+/// caller's existing 5-minute sweep is sufficient).
+pub fn cleanup_forked_peer_cooldown() {
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+    FORKED_PEER_COOLDOWN.retain(|_, marked_at| {
+        now_ms.saturating_sub(*marked_at) < FORKED_PEER_COOLDOWN_MS
+    });
+}
+
 /// Periodic cleanup of stale witness entries below `min_height`.
 /// Called by unified_p2p cleanup tasks.
 pub fn cleanup_break_tracker(min_height: u64) {
 
@@ -159,3 +159,123 @@ pub fn get_all_vrf_keys() -> HashMap<String, Vec<u8>> {
     VRF_PK_REGISTRY.read().clone()
 }
 
+// =========================================================================
+// v16.1: GENESIS DILITHIUM ANCHOR LOADER (chain-anchored identity binding)
+// =========================================================================
+//
+// Identity-key binding for the 5 genesis bootstrap nodes is anchored at boot
+// time from a JSON file shipped with the deployment. Once installed via
+// `consensus_crypto::set_genesis_anchor_pks`, the anchor map is immutable and
+// guards every subsequent registration: any PK that does not match the anchor
+// is rejected as a squat attempt.
+//
+// File format (`/app/data/genesis_anchors.json`):
+//   { "genesis_node_001": "<hex_1952_bytes>", ... "genesis_node_005": "..." }
+//
+// Operator workflow:
+//   1. On a clean cluster, every bootstrap node generates its own keypair
+//      (lazy, on first start) under `/app/data/keys/dilithium_keypair.bin`.
+//   2. Operator collects each node's PK (hex from `pk_hash` log or RPC) and
+//      writes them all into ONE JSON file deployed to every node BEFORE the
+//      first restart that loads anchors.
+//   3. Subsequent restarts read the file and install anchors at startup,
+//      BEFORE P2P comes online — closes the trust-on-first-verify race that
+//      caused the v15.x pk_mismatch deadlock.
+//
+// Operational property: keypair files MUST be backed up. If a node's
+// `dilithium_keypair.bin` is lost while the anchor map still binds the old
+// PK, the node refuses to start (via `initialize_wallet_identity`'s strict
+// guard) — operator must restore from backup.
+//
+// Scalability: anchors are bounded to the 5 genesis identities. For
+// thousands of super-node operators, identity-key binding is established via
+// signed `NodeRegistration` transactions (already implemented at
+// `cache_node_registrations_from_transactions_with_dashmap`), which carry
+// `dilithium_public_key` in TX payload and feed `register_consensus_pk_from_chain`.
+// =========================================================================
+
+/// Default location of the genesis anchors JSON file inside the container.
+pub const GENESIS_ANCHORS_PATH: &str = "/app/data/genesis_anchors.json";
+
+/// Load genesis Dilithium3 anchor PKs from `path`. Returns empty map if file
+/// missing or malformed (logged as WARN, not fatal — boot proceeds without
+/// anchors so a fresh cluster can complete first-time keygen + anchor write).
+///
+/// Format: JSON object `{ node_id: pk_hex_1952_bytes }`. Each PK MUST decode
+/// to exactly 1952 bytes; invalid entries are skipped with WARN.
+pub fn load_genesis_anchor_pks_from_file(path: &str) -> HashMap<String, Vec<u8>> {
+    use std::fs;
+    let raw = match fs::read_to_string(path) {
+        Ok(s) => s,
+        Err(_) => {
+            // Not present is normal for first cluster boot — operator writes
+            // the file after collecting PKs. Don't WARN at this stage.
+            return HashMap::new();
+        }
+    };
+
+    let parsed: HashMap<String, String> = match serde_json::from_str(&raw) {
+        Ok(m) => m,
+        Err(e) => {
+            eprintln!("[WARN][GENESIS] anchors_parse_fail path={} err={}", path, e);
+            return HashMap::new();
+        }
+    };
+
+    let mut out = HashMap::with_capacity(parsed.len());
+    for (node_id, pk_hex) in parsed {
+        match hex::decode(&pk_hex) {
+            Ok(bytes) if bytes.len() == 1952 => {
+                out.insert(node_id, bytes);
+            }
+            Ok(bytes) => {
+                eprintln!(
+                    "[WARN][GENESIS] anchor_invalid_size node={} got={} expected=1952",
+                    node_id, bytes.len()
+                );
+            }
+            Err(e) => {
+                eprintln!("[WARN][GENESIS] anchor_hex_decode_fail node={} err={}", node_id, e);
+            }
+        }
+    }
+    out
+}
+
+/// One-shot startup hook: load anchors from default path and install into
+/// the consensus-layer registry. Idempotent — second call is a no-op once
+/// the consensus layer has installed a non-empty anchor map.
+///
+/// Returns the count of anchors installed (0 if file missing — caller may
+/// proceed without anchors during first-cluster keygen, then call again
+/// after the file is written by operator).
+///
+/// MUST be called BEFORE any P2P traffic is accepted (specifically, before
+/// the first `VrfLeaderClaim` or `VrfKeyAnnounce` could trigger
+/// `register_consensus_pk_from_chain` in a TOFV path).
+pub fn install_genesis_anchors_at_startup() -> usize {
+    let map = load_genesis_anchor_pks_from_file(GENESIS_ANCHORS_PATH);
+    if map.is_empty() {
+        // First-boot path: no anchor file yet. Caller logs the appropriate
+        // INFO; we return 0 so caller can decide whether to fail or proceed.
+        return 0;
+    }
+    let count = map.len();
+    let installed = qnet_consensus::consensus_crypto::set_genesis_anchor_pks(map);
+    if installed {
+        println!("[INFO][GENESIS] anchors_installed count={} src={}", count, GENESIS_ANCHORS_PATH);
+        count
+    } else {
+        // Already installed (immutable). Treat as success, but log so an
+        // operator restart with a different file is visible.
+        println!("[INFO][GENESIS] anchors_already_installed count={}", count);
+        count
+    }
+}
+
+/// Lookup the anchored PK for a given genesis node_id. Returns None if no
+/// anchor map is installed, or if the node_id is not a genesis identity.
+pub fn get_genesis_anchor_pk(node_id: &str) -> Option<Vec<u8>> {
+    qnet_consensus::consensus_crypto::get_consensus_pk_anchor(node_id)
+}
+