consensus: slot-anchored block_ts, 2f+1-certified reward_root, 100k-scale reward+sigverify

block timestamps
- block_ts = genesis_ts + height*SLOT (deterministic, clock-independent); drop
  median-past/future/monotonic checks; slot exact-match on the live verify path
- stall pacemaker on a local wall-clock liveness timer (STALL_PROGRESS_HEIGHT/WALL),
  not slot ts, so a healthy leader is never view-changed on cumulative deficit

reward integrity (off state_root, lazy merkle-claim)
- bind per-epoch reward_root into the Checkpoint hash so the 2f+1 QC certifies it;
  committee re-derives + compares in content_ok before signing (no single-producer trust)
- thread reward_root through driver/node/window-end; finalization adopts the
  QC-certified root on every node so a catch-up node never serves a divergent local root

reward recompute scalability (100k super-nodes / <=1000 committee)
- light-bitmap index at apply (save/load_light_bitmaps): O(5) keys vs an O(14400)-block scan
- super-eligible index at apply (save/load_super_eligible): O(eligible) prefix read vs
  O(registered) per-account scan; set-equal to the old scan, survives body pruning

signature verification scalability
- shared process-wide tokio runtime for sig verify (no per-signature runtime build)
- rayon-parallel QC signature verification (committee up to ~1000 PQ sigs, no aggregation)

determinism & logging
- deterministic parallel-executor batch order (HashSet -> BTreeSet)
- sig-reject ERR->WARN, shred pacing tag ->INFO, ip-gate sampled log + always-counted metric

tests: reward_root hash-binding + light-bitmap/super-eligible index round-trip;
qnet-consensus 92 + qnet-integration 162 lib tests pass.

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

Author: AIQnetLab

Cargo.lock

@@ -59,6 +59,9 @@ dashmap = "5.5"
 parking_lot = "0.12"
 crossbeam = "0.8"
 
+# Parallel signature verification (QC committee can be ≤1000 post-quantum sigs)
+rayon = "1.10"
+
 # Metrics
 prometheus = "0.13"
 lazy_static = "1.4"

core/qnet-consensus/Cargo.toml

@@ -223,6 +223,10 @@ pub struct Checkpoint {
     /// eligible producers + committee). In the QC-signed hash ⇒ 2f+1 certify the
     /// validator set, so syncing (non-committee) nodes trust it without re-deriving.
     pub epoch_commitment: Hash,
+    /// Per-epoch reward merkle root for the emission-boundary window ([0;32] otherwise);
+    /// QC-signed ⇒ 2f+1 certify it ⇒ nodes adopt this root for claims, never a single
+    /// producer's unverified value (no Byzantine/lag reward divergence).
+    pub reward_root: Hash,
     /// Proposer's wall-clock for this window (the head microblock's timestamp).
     /// In the QC-signed hash ⇒ agreed by the committee ⇒ every node seals an
     /// identical MacroBlock from the checkpoint (no producer dependency, no fork).
@@ -246,6 +250,7 @@ impl Checkpoint {
         h.update(self.state_root);
         h.update(self.beacon);
         h.update(self.epoch_commitment);
+        h.update(self.reward_root);
         h.update(self.timestamp.to_le_bytes());
         h.update(self.proposer.as_bytes());
         h.finalize().into()
@@ -275,7 +280,7 @@ pub fn sig_merkle_root(sigs: &[Vec<u8>]) -> Hash {
 impl QuorumCertificate {
     /// Structural + cryptographic validity. `verify_sig(voter, msg, sig)` is
     /// injected so this crate stays crypto-agnostic. `committee` = sorted epoch set.
-    pub fn verify<F: Fn(&str, &[u8], &[u8]) -> bool>(
+    pub fn verify<F: Fn(&str, &[u8], &[u8]) -> bool + Sync>(
         &self,
         committee: &[NodeId],
         verify_sig: F,
@@ -289,9 +294,13 @@ impl QuorumCertificate {
             if !committee.iter().any(|c| c == s) { return Err("qc_non_member"); }
         }
         if sig_merkle_root(&self.sigs) != self.sig_merkle_root { return Err("qc_merkle_mismatch"); }
-        for (voter, sig) in self.signers.iter().zip(self.sigs.iter()) {
-            if !verify_sig(voter, &self.checkpoint_hash, sig) { return Err("qc_bad_sig"); }
-        }
+        // Verify signatures in parallel: a committee is ≤1000 post-quantum sigs and ML-DSA has no
+        // BLS-style aggregation, so par_iter spreads the per-sig Dilithium open across cores.
+        // all() ≡ the serial AND (short-circuits on the first invalid); order is irrelevant.
+        use rayon::prelude::*;
+        let all_ok = self.signers.par_iter().zip(self.sigs.par_iter())
+            .all(|(voter, sig)| verify_sig(voter, &self.checkpoint_hash, sig));
+        if !all_ok { return Err("qc_bad_sig"); }
         Ok(())
     }
 }
@@ -411,13 +420,21 @@ mod tests {
         let mut c = Checkpoint {
             index: 1, parent_qc: None, window_head_height: 90,
             window_mb_hashes: vec![h(1), h(2)], state_root: h(3),
-            beacon: h(4), epoch_commitment: h(0), timestamp: 0, proposer: "n1".into(), proposer_sig: vec![1,2,3],
+            beacon: h(4), epoch_commitment: h(0), reward_root: h(0), timestamp: 0, proposer: "n1".into(), proposer_sig: vec![1,2,3],
         };
         let x = c.hash();
         c.proposer_sig = vec![9, 9, 9];   // sig change must NOT change hash
         assert_eq!(c.hash(), x);
         c.state_root = h(7);              // field change MUST change hash
         assert_ne!(c.hash(), x);
+        // reward_root MUST be bound into the hash: the QC signs cp.hash(), so a checkpoint
+        // differing only in reward_root must produce a different hash (else 2f+1 could not
+        // certify the reward distribution and a proposer-chosen root would ride uncertified).
+        let mut a = c.clone();
+        a.reward_root = h(0);
+        let mut b = a.clone();
+        b.reward_root = h(5);
+        assert_ne!(a.hash(), b.hash());
     }
 
     #[test]
@@ -585,7 +602,7 @@ mod tests {
         let child = Checkpoint {
             index: 5, parent_qc: Some(parent_qc), window_head_height: 450,
             window_mb_hashes: vec![h(1)], state_root: h(2), beacon: h(3),
-            epoch_commitment: h(0), timestamp: 0, proposer: "n0".into(), proposer_sig: vec![],
+            epoch_commitment: h(0), reward_root: h(0), timestamp: 0, proposer: "n0".into(), proposer_sig: vec![],
         };
         let child_qc = mk_qc(&committee, child.hash(), 5, 3);
         assert_eq!(commits_parent(&child, &child_qc), Some(4)); // C4 final
@@ -601,7 +618,7 @@ mod tests {
         let c = Checkpoint {
             index: 7, parent_qc: Some(qc.clone()), window_head_height: 630,
             window_mb_hashes: vec![h(1), h(2)], state_root: h(3), beacon: h(4),
-            epoch_commitment: h(0), timestamp: 0, proposer: "n1".into(), proposer_sig: vec![1,2,3],
+            epoch_commitment: h(0), reward_root: h(0), timestamp: 0, proposer: "n1".into(), proposer_sig: vec![1,2,3],
         };
         let bytes = bincode::serialize(&c).unwrap();
         let back: Checkpoint = bincode::deserialize(&bytes).unwrap();

core/qnet-consensus/src/checkpoint_bft.rs

@@ -229,7 +229,7 @@ mod tests {
         Checkpoint {
             index, parent_qc, window_head_height: index * 10,
             window_mb_hashes: vec![hh(index as u8)], state_root: hh(index as u8),
-            beacon: hh(0), epoch_commitment: hh(0), timestamp: 0, proposer: c[li].clone(), proposer_sig: Vec::new(),
+            beacon: hh(0), epoch_commitment: hh(0), reward_root: hh(0), timestamp: 0, proposer: c[li].clone(), proposer_sig: Vec::new(),
         }
     }
 

core/qnet-consensus/src/checkpoint_consensus.rs

@@ -2081,8 +2081,9 @@ async fn verify_with_real_dilithium(
                          node_id, hex::encode(&public_key_bytes[..8]));
                 return true;
             } else {
-                eprintln!("[ERR][CONSENSUS] msg_mismatch node={} expected_len={} recovered_len={}",
-                         node_id, expected_msg.len(), recovered_message.len());
+                // Recovered != expected: stale/cross-round signature or forgery. Rejection
+                // is unconditional (security boundary); expected at low rate ⇒ WARN, not ERR.
+                eprintln!("[WARN][CONSENSUS] sig_reject node={} reason=msg_mismatch", node_id);
                 return false;
             }
         }

core/qnet-consensus/src/consensus_crypto.rs

@@ -1795,44 +1795,19 @@ impl BlockPipeline {
                 }
             }
 
-            // 2. Timestamp validation (only in live mode — sync mode skips)
-            // v14.8.11: three-check canonical timestamp model.
-            //   (a) FUTURE:       block.ts ≤ wall_clock + TIMESTAMP_FUTURE_TOLERANCE
-            //   (b) MEDIAN-PAST:  block.ts > median(last 11 on-chain timestamps)
-            //   (c) MONOTONICITY: block.ts > parent.ts (enforced in step 1 hash chain)
-            // Same ruleset is applied in `validate_received_microblock`; this
-            // pipeline check catches bad blocks before they reach the apply
-            // stage.
+            // 2. Slot-anchored timestamp validation (LIVE only; SYNC skips — block_ts is
+            // already bound by the block hash + producer Dilithium sig + hash-chain).
+            // block_ts must equal genesis_ts + height*SLOT exactly: deterministic,
+            // clock-independent, non-gameable. The single source of truth on the live path.
             let snap = coordinator.snapshot();
             if !snap.is_syncing() && mb.height > 0 {
-                let now = crate::node::get_timestamp_safe();
-                // (a) FUTURE check against raw wall clock: canonical design, so
-                // an attacker cannot game the window via network-adjusted time.
-                // TIMESTAMP_FUTURE_TOLERANCE = 7200 s (2 h) comfortably covers
-                // realistic hypervisor / NTP events without letting Byzantine
-                // inflation go unchecked.
-                if mb.timestamp > now + crate::node::TIMESTAMP_FUTURE_TOLERANCE {
-                    if is_warn() {
-                        println!("[WARN][PIPELINE] future_block h={} delta=+{}s from={}",
-                                 mb.height, mb.timestamp.saturating_sub(now), decoded.from_peer);
-                    }
-                    metrics.verify_failed.fetch_add(1, Ordering::Relaxed);
-                    continue;
-                }
-                // (b) MEDIAN-PAST rule. Silently skipped during the first ~11
-                // blocks after boot when the ring is undersized.
-                // v15.15: height-ordered (BIP-113 canonical) — median is taken
-                // over the parent-chain segment [mb.height - WINDOW, mb.height),
-                // not over the last WINDOW insertions. Fixes false positives
-                // during catch-up sync where blocks legitimately re-arrive in
-                // non-height order.
-                if let Some(median_past) = crate::node::median_past_timestamp_at_height(mb.height) {
-                    if mb.timestamp <= median_past {
+                let g = crate::GLOBAL_GENESIS_TIMESTAMP.load(Ordering::Relaxed);
+                if g != 0 {
+                    let expected = crate::node::expected_block_timestamp(g, mb.height);
+                    if mb.timestamp != expected {
                         if is_warn() {
-                            println!(
-                                "[WARN][PIPELINE] median_past_violation h={} ts={} median_past={} from={}",
-                                mb.height, mb.timestamp, median_past, decoded.from_peer
-                            );
+                            println!("[WARN][PIPELINE] slot_mismatch h={} ts={} expected={} from={}",
+                                     mb.height, mb.timestamp, expected, decoded.from_peer);
                         }
                         metrics.verify_failed.fetch_add(1, Ordering::Relaxed);
                         continue;

development/qnet-integration/src/block_pipeline.rs

@@ -121,7 +121,7 @@ impl ConsensusDriver {
     pub fn build_proposal(
         &mut self, window: u64, mb_hashes: Vec<Hash>,
         state_root: Hash, beacon: Hash, head_ts: u64,
-        committee: Vec<NodeId>, eligible_producers: Vec<u8>, banned: Vec<NodeId>,
+        committee: Vec<NodeId>, eligible_producers: Vec<u8>, banned: Vec<NodeId>, reward_root: Hash,
     ) -> Vec<Effect> {
         self.set_committee(committee.clone());
         let round = self.eng.current_index;
@@ -139,7 +139,7 @@ impl ConsensusDriver {
         let head_height = window.saturating_mul(self.cp_interval);
         let cp = Checkpoint {
             index: round, parent_qc: self.eng.high_qc.clone(), window_head_height: head_height,
-            window_mb_hashes: mb_hashes, state_root, beacon, epoch_commitment: epoch_c, timestamp: head_ts,
+            window_mb_hashes: mb_hashes, state_root, beacon, epoch_commitment: epoch_c, reward_root, timestamp: head_ts,
             proposer: self.node_id.clone(), proposer_sig: Vec::new(),
         };
         self.last_proposed_round = round;
@@ -312,7 +312,7 @@ mod tests {
         for index in 1..=8u64 {
             let mut seed = Vec::new();
             for k in 0..nodes.len() {
-                let effs = nodes[k].d.build_proposal(index, vec![[index as u8; 32]], [index as u8; 32], [0u8; 32], index * 1000, c.clone(), Vec::new(), Vec::new());
+                let effs = nodes[k].d.build_proposal(index, vec![[index as u8; 32]], [index as u8; 32], [0u8; 32], index * 1000, c.clone(), Vec::new(), Vec::new(), [0u8; 32]);
                 for e in effs { seed.extend(exec(&mut nodes[k], e)); }
             }
             deliver(&mut nodes, &c, seed);
@@ -340,7 +340,7 @@ mod tests {
         for index in 1..=6u64 {
             let mut seed = Vec::new();
             for k in 0..nodes.len() {
-                let effs = nodes[k].d.build_proposal(index, vec![[index as u8; 32]], [index as u8; 32], [0u8; 32], index * 1000, c.clone(), Vec::new(), Vec::new());
+                let effs = nodes[k].d.build_proposal(index, vec![[index as u8; 32]], [index as u8; 32], [0u8; 32], index * 1000, c.clone(), Vec::new(), Vec::new(), [0u8; 32]);
                 for e in effs { seed.extend(exec(&mut nodes[k], e)); }
             }
             deliver(&mut nodes, &c, seed);
@@ -359,7 +359,7 @@ mod tests {
         let c: Vec<NodeId> = (0..4).map(|i| format!("n{}", i)).collect();
         let cp = Checkpoint {
             index: 1, parent_qc: None, window_head_height: 10, window_mb_hashes: vec![[1u8; 32]],
-            state_root: [1u8; 32], beacon: [0u8; 32], epoch_commitment: [0u8; 32], timestamp: 0, proposer: "n1".into(), proposer_sig: vec![9, 9],
+            state_root: [1u8; 32], beacon: [0u8; 32], epoch_commitment: [0u8; 32], reward_root: [0u8; 32], timestamp: 0, proposer: "n1".into(), proposer_sig: vec![9, 9],
         };
         // forged proposer_sig fails node verify ⇒ never reaches the driver
         assert!(!verify_msg(&c, &ConsensusMsg::Proposal(cp)));
@@ -382,7 +382,7 @@ mod tests {
         fn step(nodes: &mut Vec<Node>, c: &[NodeId], w: u64) {
             let mut seed = Vec::new();
             for k in 0..nodes.len() {
-                let effs = nodes[k].d.build_proposal(w, vec![[w as u8; 32]], [w as u8; 32], [0u8; 32], w * 1000, c.to_vec(), Vec::new(), Vec::new());
+                let effs = nodes[k].d.build_proposal(w, vec![[w as u8; 32]], [w as u8; 32], [0u8; 32], w * 1000, c.to_vec(), Vec::new(), Vec::new(), [0u8; 32]);
                 for e in effs { seed.extend(exec(&mut nodes[k], e)); }
             }
             deliver(nodes, c, seed);
@@ -426,7 +426,7 @@ mod tests {
             let cp = Checkpoint {
                 index: i, parent_qc: prev_qc.clone(), window_head_height: i * 90,
                 window_mb_hashes: vec![[i as u8; 32]], state_root: [i as u8; 32],
-                beacon: [0u8; 32], epoch_commitment: [0u8; 32], timestamp: 0,
+                beacon: [0u8; 32], epoch_commitment: [0u8; 32], reward_root: [0u8; 32], timestamp: 0,
                 proposer: c[leader_index(i, &parent_hash, c.len())].clone(), proposer_sig: Vec::new(),
             };
             let signers: Vec<NodeId> = c.iter().take(3).cloned().collect();