scale: fix super eligibility epoch-boundary flicker + light-ping/FSM/apply-stuck hardening

- super eligibility: recency prev = cur-1 spans the epoch boundary; the per-epoch reset
  ejected the whole non-genesis eligible set for the first subwindow of every epoch (a
  deterministic flicker at >120 supers). Pure fn of scan_end; boundary test updated.
- light-ping selection: per-window slot-index cache, rebuilt only on window/size change;
  O(bucket)/tick instead of an O(N log N) clone+sort of the whole registry every 60s tick.
- light liveness FSM: persist is_active in a separate metadata key (off the registry_root
  row, zero consensus impact); a genesis restart no longer resurrects dropped nodes.
- pipeline watchdog: 0-sentinel progress clock so the genesis-boot wait cannot emit a
  spurious apply_stuck CRIT (no fleet-wide false-CRIT storm on coordinated restart).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: AIQnetLab

development/qnet-integration/src/block_pipeline.rs

@@ -1253,8 +1253,10 @@ impl BlockPipeline {
             const STUCK_THRESHOLD_MS: u64 = 30_000;
             let mut last_verified: u64 = 0;
             let mut last_applied: u64 = 0;
-            let mut last_verified_progress_ms: u64 = now_ms();
-            let mut last_applied_progress_ms: u64 = now_ms();
+            // 0 sentinel = "no verify/apply seen yet"; the dump guards require != 0, so the boot wait
+            // (nothing to apply) can't trip a spurious CRIT — stall is measured from first real progress.
+            let mut last_verified_progress_ms: u64 = 0;
+            let mut last_applied_progress_ms: u64 = 0;
             let mut last_verify_dump_ms: u64 = 0;
             let mut last_apply_dump_ms: u64 = 0;
             let mut interval = tokio::time::interval(WATCHDOG_TICK);
@@ -1287,6 +1289,7 @@ impl BlockPipeline {
 
                 // VERIFY STALL DUMP: counter unchanged for ≥30 s and op != idle.
                 if verify_stall_ms >= STUCK_THRESHOLD_MS
+                    && last_verified_progress_ms != 0
                     && verify_op != PIPELINE_OP_IDLE
                     && now.saturating_sub(last_verify_dump_ms) >= STUCK_THRESHOLD_MS
                 {
@@ -1309,6 +1312,7 @@ impl BlockPipeline {
 
                 // APPLY STALL DUMP: counter unchanged for ≥30 s and op != idle.
                 if apply_stall_ms >= STUCK_THRESHOLD_MS
+                    && last_applied_progress_ms != 0
                     && apply_op != PIPELINE_OP_IDLE
                     && now.saturating_sub(last_apply_dump_ms) >= STUCK_THRESHOLD_MS
                 {

development/qnet-integration/src/node.rs

@@ -912,15 +912,13 @@ pub fn window_content_from_accum(mb_idx: u64) -> Option<(Vec<[u8; 32]>, Vec<[u8;
 /// recency) and that was included in a block at-or-below scan_end. A pure function of the canonical chain
 /// ≤ scan_end ⇒ identical on every committee member, with NO live-tip dependence. Scans ≤2 subwindows
 /// (~2880 blocks), off the production path; bodies are retained 6 epochs, far beyond this window.
-/// (current, previous) global subwindow indices (anchor/1440 = epoch*10 + subwindow) for `scan_end`. The
-/// previous counts ONLY within the SAME epoch — at a subwindow-0/epoch boundary it returns current==prev
-/// (no previous), mirroring the prior `heartbeat_epoch==hb_epoch` + `if cur_sub>0` bitmask recency. Pure
-/// ⇒ unit-tested for the off-by-one + cross-epoch boundary.
+/// (current, previous) GLOBAL subwindow indices (anchor/1440) for `scan_end`; prev = cur-1 SPANS the
+/// epoch boundary — a Heartbeat from the prior epoch's last subwindow is still recent liveness. The old
+/// per-epoch reset (prev=cur at subwindow 0) ejected the whole non-genesis eligible set for the first
+/// subwindow of every epoch (the eligibility flicker at scale). Pure fn of scan_end ⇒ deterministic.
 fn recency_subwindow_indices(scan_end: u64) -> (u64, u64) {
     let cur_idx = scan_end / 1440;
-    let cur_sub = (scan_end % 14400) / 1440; // 0..9 within the epoch
-    let prev_idx = if cur_sub > 0 { cur_idx.saturating_sub(1) } else { cur_idx };
-    (cur_idx, prev_idx)
+    (cur_idx, cur_idx.saturating_sub(1))
 }
 
 fn recent_heartbeat_senders(storage: &crate::storage::Storage, scan_end: u64) -> std::collections::HashSet<String> {
@@ -27610,22 +27608,23 @@ mod tests {
         assert!(!checkpoint_participation_allowed(false, 0, mb_end));      // syncing, no window → defer
     }
 
-    // Phase-2A recency window (deterministic heartbeat eligibility): current subwindow + previous ONLY
-    // within the same epoch. The epoch boundary (subwindow 0) must NOT bridge to the prior epoch's
-    // subwindow 9 — that is the off-by-one that would change WHO is eligible vs the old bitmask gate.
+    // Phase-2A recency window: current subwindow + the immediately previous GLOBAL subwindow. prev = cur-1
+    // SPANS the epoch boundary (prior epoch's subwindow 9 is still recent liveness); the old per-epoch reset
+    // (prev=cur at subwindow 0) ejected the whole non-genesis eligible set each epoch start (the flicker).
     #[test]
     fn recency_subwindow_indices_boundary() {
-        // Mid-epoch: previous = current-1 (same epoch).
+        // Mid-epoch: previous = current-1.
         assert_eq!(recency_subwindow_indices(5 * 1440), (5, 4));
         assert_eq!(recency_subwindow_indices(5 * 1440 + 100), (5, 4));
-        // Epoch start (subwindow 0): no previous within the epoch ⇒ prev == cur (degenerates to {cur}).
+        // h0: no previous ⇒ saturating to (0, 0).
         assert_eq!(recency_subwindow_indices(0), (0, 0));
-        assert_eq!(recency_subwindow_indices(14400), (10, 10));        // epoch1 sub0 ⇒ (10,10), NOT (10,9)
-        assert_eq!(recency_subwindow_indices(14400 + 50), (10, 10));
-        // Epoch1 subwindow 1: previous is sub0 of the SAME epoch (10), never the prior epoch's sub9 (9).
+        // Epoch boundary (subwindow 0): prev = cur-1 BRIDGES to the prior epoch's subwindow 9 (no flicker).
+        assert_eq!(recency_subwindow_indices(14400), (10, 9));
+        assert_eq!(recency_subwindow_indices(14400 + 50), (10, 9));
+        // Epoch1 subwindow 1: previous is subwindow 0 of the same epoch.
         assert_eq!(recency_subwindow_indices(14400 + 1440), (11, 10));
-        // Epoch2 subwindow 0: again no bridge.
-        assert_eq!(recency_subwindow_indices(2 * 14400), (20, 20));
+        // Epoch2 boundary: bridges to epoch1 subwindow 9.
+        assert_eq!(recency_subwindow_indices(2 * 14400), (20, 19));
     }
 
     // committee_for_height determinism: genesis era ⇒ None (caller uses the genesis committee), and

development/qnet-integration/src/storage.rs

@@ -6645,6 +6645,25 @@ impl Storage {
         self.save_node_registration_inner(node_id, node_type, wallet, reputation, Some(reg_height), Some(burn_tx), vrf_pk)
     }
 
+    /// Light-node liveness FSM persistence (durable across genesis restart). A separate metadata-CF key,
+    /// NOT the registry_root-committed node_ row ⇒ zero consensus impact. Written only on the is_active
+    /// FLIP (drop-after-5-misses / reactivate), so a restart does not silently resurrect a dropped node.
+    pub fn mark_light_inactive(&self, node_id: &str) {
+        if let Some(cf) = self.persistent.db.cf_handle("metadata") {
+            let _ = self.persistent.db.put_cf(&cf, format!("lninact_{}", node_id).as_bytes(), b"1");
+        }
+    }
+    pub fn clear_light_inactive(&self, node_id: &str) {
+        if let Some(cf) = self.persistent.db.cf_handle("metadata") {
+            let _ = self.persistent.db.delete_cf(&cf, format!("lninact_{}", node_id).as_bytes());
+        }
+    }
+    pub fn is_light_inactive(&self, node_id: &str) -> bool {
+        self.persistent.db.cf_handle("metadata")
+            .and_then(|cf| self.persistent.db.get_cf(&cf, format!("lninact_{}", node_id).as_bytes()).ok().flatten())
+            .is_some()
+    }
+
     fn save_node_registration_inner(&self, node_id: &str, node_type: &str, wallet: &str, reputation: f64, reg_height: Option<u64>, burn_tx: Option<&str>, vrf_pk: Option<&[u8]>) -> IntegrationResult<()> {
         let registry_cf = self.persistent.db.cf_handle("node_registry")
             .ok_or_else(|| IntegrationError::StorageError("node_registry column family not found".to_string()))?;

development/qnet-integration/src/unified_p2p.rs

@@ -2155,8 +2155,12 @@ pub struct SimplifiedP2P {
     /// PRODUCTION: Light Node registry synchronized via gossip
     /// All Super nodes maintain identical registry for deterministic ping assignment
     light_node_registry: Arc<RwLock<HashMap<String, LightNodeRegistrationData>>>,
-    
-    
+    /// Per-window ping-slot buckets for THIS genesis's shard: (window, registry_len, [slot 0..239]→node_ids).
+    /// Rebuilt only on window/size change ⇒ ping selection is O(bucket)/tick, not O(N log N) clone+sort of
+    /// the whole registry every tick — scales the 5-genesis pinger to millions of light nodes.
+    light_ping_slot_cache: Arc<RwLock<(u64, usize, Vec<Vec<String>>)>>,
+
+
     /// PRODUCTION: Storage reference for persistent heartbeat storage
     /// SCALABILITY: Each node stores ONLY its own heartbeats in RocksDB (10 records per 4h)
     /// Supports millions of nodes without RAM limitations
@@ -3071,7 +3075,8 @@ impl SimplifiedP2P {
 
             // PRODUCTION: Light Node registry for gossip sync
             light_node_registry: Arc::new(RwLock::new(HashMap::new())),
-            
+            light_ping_slot_cache: Arc::new(RwLock::new((u64::MAX, 0, Vec::new()))),
+
             // PRODUCTION: Heartbeat history for reward eligibility
             storage: storage, // v2.76: Storage for persistent heartbeat storage
 
@@ -16964,6 +16969,9 @@ impl SimplifiedP2P {
 
         for (node_id, wallet_address, _node_type, registered_at) in nodes {
             if !registry.contains_key(&node_id) {
+                // Restore the persisted liveness FSM instead of resurrecting to active — a dropped node
+                // stays dropped across a genesis restart until it re-attests (honors manual reactivation).
+                let inactive = self.storage.as_ref().map(|s| s.is_light_inactive(&node_id)).unwrap_or(false);
                 registry.insert(node_id.clone(), LightNodeRegistrationData {
                     node_id,
                     wallet_address,
@@ -16974,8 +16982,8 @@ impl SimplifiedP2P {
                     push_type: PushType::Polling,
                     unified_push_endpoint: None,
                     last_seen: registered_at,
-                    consecutive_failures: 0,
-                    is_active: true,
+                    consecutive_failures: if inactive { 5 } else { 0 },
+                    is_active: !inactive,
                     ed25519_signature: String::new(),
                     ed25519_public_key: String::new(),
                     ping_pubkey: String::new(),         // Populated on re-registration
@@ -17251,95 +17259,89 @@ impl SimplifiedP2P {
     /// SCALABILITY: 2M pings per Genesis per epoch = 139 pings/sec = easily handled
     pub fn get_light_nodes_to_ping(&self) -> Vec<(LightNodeRegistrationData, PingerRole)> {
         let current_slot = Self::get_current_slot();
+        let current_window = Self::get_current_window_number();
         let our_node_id = &self.node_id;
         let mut result = Vec::new();
-        
-        // v2.89: ONLY Genesis nodes can ping Light nodes
-        // This ensures 100% reliability - Genesis never goes offline
+
+        // v2.89: ONLY Genesis nodes ping Light nodes (5 fixed shard owners, always online).
         let is_genesis_node = std::env::var("QNET_BOOTSTRAP_ID")
             .map(|id| ["001", "002", "003", "004", "005"].contains(&id.as_str()))
             .unwrap_or(false);
-        
-        if !is_genesis_node {
-            // Non-Genesis nodes don't ping Light nodes anymore
-            // This prevents data loss when regular nodes go offline
-            return result;
-        }
-        
-        // Get our Genesis index (0-4) for shard assignment
+        if !is_genesis_node { return result; }
         let our_genesis_idx = std::env::var("QNET_BOOTSTRAP_ID")
-            .ok()
-            .and_then(|id| id.parse::<usize>().ok())
-            .map(|id| id.saturating_sub(1)) // Convert 001-005 to 0-4
-            .unwrap_or(0);
-        
+            .ok().and_then(|id| id.parse::<usize>().ok())
+            .map(|id| id.saturating_sub(1)).unwrap_or(0);
         const GENESIS_COUNT: usize = 5;
-        
+
         if crate::node::is_info() {
             println!("[INFO][GENESIS-PING] Genesis node {} (idx={}) checking Light nodes to ping slot={}",
                      our_node_id, our_genesis_idx, current_slot);
         }
-        
-        // Get all Light nodes from registry SORTED for consistent linear sharding
-        // v2.89 CRITICAL: Must use same ordering as bitmap creation!
+
         let registry = self.light_node_registry.read();
-        let mut all_nodes: Vec<_> = registry.values().cloned().collect();
-        all_nodes.sort_by(|a, b| a.node_id.cmp(&b.node_id)); // Sort by node_id
-        let total_light_nodes = all_nodes.len();
-        
-        // v2.89: LINEAR SHARDING - each Genesis gets sequential range of indices
-        // This matches bitmap creation logic exactly!
-        let nodes_per_genesis = (total_light_nodes + GENESIS_COUNT - 1) / GENESIS_COUNT; // Ceiling division
-        let my_start = our_genesis_idx * nodes_per_genesis;
-        let my_end = std::cmp::min(my_start + nodes_per_genesis, total_light_nodes);
-        
-        for idx in my_start..my_end {
-            let node = &all_nodes[idx];
-            
-            // ACTIVITY FILTER: Skip inactive nodes (>5 consecutive failures)
-            if !node.is_active || node.consecutive_failures >= 5 {
-                continue;
-            }
-            
-            // Check if this is the node's ping slot (randomized per window)
-            if !Self::is_light_node_ping_slot(&node.node_id) {
-                continue;
+        let reg_len = registry.len();
+
+        // Rebuild this genesis's per-slot buckets only when the window rolls or the registry size changes.
+        // Linear sharding over the SORTED registry (must match bitmap creation). O(N log N) once per window,
+        // not every 60s tick — the O(N log N) clone+sort per tick did not scale to millions of light nodes.
+        let need_rebuild = { let c = self.light_ping_slot_cache.read(); c.0 != current_window || c.1 != reg_len };
+        if need_rebuild {
+            let mut ids: Vec<&String> = registry.keys().collect();
+            ids.sort();
+            let nodes_per_genesis = (reg_len + GENESIS_COUNT - 1) / GENESIS_COUNT;
+            let my_start = our_genesis_idx * nodes_per_genesis;
+            let my_end = std::cmp::min(my_start + nodes_per_genesis, reg_len);
+            let mut buckets: Vec<Vec<String>> = vec![Vec::new(); 240];
+            for i in my_start..my_end {
+                let id = ids[i];
+                let slot = Self::calculate_randomized_slot(id, current_window) as usize;
+                buckets[slot].push(id.clone());
+            }
+            *self.light_ping_slot_cache.write() = (current_window, reg_len, buckets);
+        }
+
+        // Read the 3 grace slots {cur, cur-1, cur-2} (mod 240) — same window as is_light_node_ping_slot
+        // (slot_diff<=2). is_active/consecutive_failures/attested are mutable ⇒ filtered per tick.
+        let cache = self.light_ping_slot_cache.read();
+        for g in 0..=2u64 {
+            let s = ((current_slot + 240 - g) % 240) as usize;
+            for node_id in cache.2.get(s).into_iter().flatten() {
+                let node = match registry.get(node_id) { Some(n) => n, None => continue };
+                if !node.is_active || node.consecutive_failures >= 5 { continue; }
+                if self.has_attestation_in_window(node_id) { continue; }
+                result.push((node.clone(), PingerRole::Primary));
             }
-            
-            // Skip if already attested in ANY slot of the current 4h window.
-            // The 3-slot retry window is only for failures — if the node already proved
-            // liveness this window, no further FCM notifications are needed.
-            if self.has_attestation_in_window(&node.node_id) {
-                continue;
-            }
-            
-            // Genesis is always Primary pinger (no backup needed - Genesis is reliable)
-            result.push((node.clone(), PingerRole::Primary));
         }
-        
+
         if crate::node::is_debug() && !result.is_empty() {
-            println!("[DBG][GENESIS-PING] Genesis {} has {} Light nodes to ping this slot (total registry: {})",
-                     our_genesis_idx + 1, result.len(), total_light_nodes);
+            println!("[DBG][GENESIS-PING] Genesis {} has {} Light nodes to ping this slot (registry: {})",
+                     our_genesis_idx + 1, result.len(), reg_len);
         }
-        
         result
     }
 
     /// Mark Light node as failed (no response to ping)
     /// After 5 consecutive failures, node is marked inactive
     pub fn mark_light_node_ping_failed(&self, node_id: &str) {
-        let mut registry = self.light_node_registry.write();
-        if let Some(node) = registry.get_mut(node_id) {
-            node.consecutive_failures = node.consecutive_failures.saturating_add(1);
-            
-            if node.consecutive_failures >= 5 {
-                node.is_active = false;
-                if crate::node::is_info() {
-                    println!("[WARN][P2P] Node {} marked inactive after {} consecutive failures",
-                             node_id, node.consecutive_failures);
+        let mut flipped_inactive = false;
+        {
+            let mut registry = self.light_node_registry.write();
+            if let Some(node) = registry.get_mut(node_id) {
+                node.consecutive_failures = node.consecutive_failures.saturating_add(1);
+                if node.consecutive_failures >= 5 && node.is_active {
+                    node.is_active = false;
+                    flipped_inactive = true;
+                    if crate::node::is_info() {
+                        println!("[WARN][P2P] Node {} marked inactive after {} consecutive failures",
+                                 node_id, node.consecutive_failures);
+                    }
                 }
             }
         }
+        // Persist the drop so a genesis restart does not silently resurrect it (honors manual reactivation).
+        if flipped_inactive {
+            if let Some(s) = &self.storage { s.mark_light_inactive(node_id); }
+        }
     }
 
     /// Update push_type + last_seen for a light node (called on token-refresh).
@@ -17372,6 +17374,7 @@ impl SimplifiedP2P {
             node.is_active = true;
 
             if was_inactive {
+                if let Some(s) = &self.storage { s.clear_light_inactive(node_id); }
                 if crate::node::is_info() {
                     println!("[INFO][P2P] Node {} reactivated after successful ping", node_id);
                 }