fix: cold-join fast-sync binding + reliable NodeRegistration delivery

D2 — snapshot fast-sync (Pattern C) now actually binds (was never wired):
- boundary snapshot is the canonical full_snap_ accounts-CF dump on every
  node's apply path (was a broken in-memory bincode state_snap_ that restored
  to garbage, so binding always failed and joiners fell to O(N) block replay)
- verifier binds restored state to the QC-certified macroblock state_root:
  rebuild finalize_merkle from the restored accounts and compare to
  mb.state_root, instead of the never-assigned consensus_data.snapshot_root
- fetch the macroblock prefix [latest+1..=mb_idx] so the binding macroblock
  persists (N-2 anchor) and is QC-validated by the normal store path

D3 — NodeRegistration lands like NodeActivation:
- boot send + periodic rebroadcast fan out the binding TX to all genesis nodes
  (was a single best-effort gossip with no fan-out)
- rebroadcast runs sync-independently; producer eligibility stays gated at
  selection, not at TX broadcast
- on boot, re-send unless the registration is already on-chain (chain is the
  source of truth, not the local activation code)
- mobile/server-side registration path gets the same genesis fan-out

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

Author: AIQnetLab

development/qnet-integration/src/bin/qnet-node.rs

@@ -2880,8 +2880,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let already_persisted = node.get_storage().load_activation_code()
             .map(|opt| opt.is_some())
             .unwrap_or(false);
-        if already_persisted {
-            if is_info() { println!("[INFO][NODE] activation_already_persisted skip=fallback_call"); }
+        // Chain is the source of truth: skip the (re)send only if the code is persisted AND this
+        // node's NodeRegistration is already on-chain. If persisted on an earlier boot but the
+        // registration never landed (dropped join-time broadcast), re-send so the binding TX reaches
+        // a producer. One send per boot; the early-activation path is sync-gated (OFF on a cold
+        // joiner) so there is no same-boot double-submit.
+        let reg_onchain = node.get_storage().is_node_registration_onchain(&node.get_node_id());
+        if already_persisted && reg_onchain {
+            if is_info() { println!("[INFO][NODE] activation_persisted reg_onchain=true skip=fallback_call"); }
         } else if let Err(e) = node.save_activation_code(&activation_code, node_type).await {
             if is_warn() { println!("[WARN][NODE] activation_code_save_failed err={}", e); }
         }

development/qnet-integration/src/block_pipeline.rs

@@ -2904,105 +2904,19 @@ impl BlockPipeline {
                 }
             }
 
-            // v15.9: committee-wide canonical snapshot at each
-            // SNAPSHOT_INCREMENTAL_INTERVAL — fresh nodes parallel-download from any
-            // committee member; macroblock snapshot_root has a byte-identical
-            // artefact on every honest node; rollback finds a snapshot <= target.
-            // SOURCE = in-memory state.accounts (the Arc<DashMap> every apply
-            // mutates), NOT the RocksDB accounts CF (only written on restore →
-            // stale/empty → bad snapshot_root). Sort by address before bincode
-            // (DashMap iter order is shard/node-dependent) → identical bytes → root
-            // converges. Off-reactor (blocking pool, Arc to accounts); pipeline
-            // returns immediately; failure WARN-only, never blocks liveness.
+            // Canonical boundary snapshot (raw accounts-CF dump via create_state_snapshot) at every
+            // SNAPSHOT_INCREMENTAL_INTERVAL, on EVERY node's apply path (not just the producer) so a
+            // cold joiner can fast-sync from any peer. This is the SAME representation the macroblock
+            // snapshot_root binds (compute_canonical_state_root over the accounts CF) → restore
+            // reproduces the bound root. Off-reactor; WARN-only, never blocks liveness.
             const SNAPSHOT_INCREMENTAL_INTERVAL: u64 = 3_600;
             if height > 0 && height % SNAPSHOT_INCREMENTAL_INTERVAL == 0 {
                 let storage_for_snapshot = ctx.storage.clone();
-                let state_for_snapshot = ctx.state.clone();
                 let snapshot_height = height;
                 tokio::spawn(async move {
-                    let start = std::time::Instant::now();
-
-                    // Read the in-memory state under a brief read lock to
-                    // capture: (a) a strong handle to the accounts map,
-                    // (b) the current state_root, (c) the current
-                    // total_supply. We drop the lock before the heavy
-                    // serialise step so block apply is not blocked.
-                    let (accounts_arc, state_root, total_supply) = {
-                        let sg = state_for_snapshot.read().await;
-                        let accounts_arc = sg.accounts.clone();
-                        let state_root = sg.calculate_state_root().unwrap_or([0u8; 32]);
-                        let total_supply = sg.chain_state.read().total_supply;
-                        (accounts_arc, state_root, total_supply)
-                    };
-
-                    // Heavy work: iterate DashMap, clone, sort, bincode.
-                    // Lives on the blocking thread pool so the reactor
-                    // stays free; the closure consumes `accounts_arc` so
-                    // no shared-state hazards remain after spawn.
-                    let serialise_result = tokio::task::spawn_blocking(move || {
-                        let mut accounts: Vec<(String, qnet_state::Account)> = accounts_arc
-                            .iter()
-                            .map(|e| (e.key().clone(), e.value().clone()))
-                            .collect();
-                        accounts.sort_by(|a, b| a.0.cmp(&b.0));
-                        bincode::serialize(&accounts)
-                    }).await;
-
-                    let state_data = match serialise_result {
-                        Ok(Ok(data)) => data,
-                        Ok(Err(e)) => {
-                            if is_warn() {
-                                println!(
-                                    "[WARN][PIPELINE] snapshot_serialize_fail h={} err={}",
-                                    snapshot_height, e,
-                                );
-                            }
-                            return;
-                        }
-                        Err(e) => {
-                            if is_warn() {
-                                println!(
-                                    "[WARN][PIPELINE] snapshot_join_fail h={} err={:?}",
-                                    snapshot_height, e,
-                                );
-                            }
-                            return;
-                        }
-                    };
-
-                    if state_data.is_empty() {
-                        // Genesis-window or pre-state node — nothing to bind.
-                        return;
-                    }
-
-                    // Write the canonical snapshot artefact. `save_state_snapshot`
-                    // wraps zstd-15 + integrity hash + atomic batch write —
-                    // already off-reactor (Fix #2 spawn_blocking).
-                    match storage_for_snapshot
-                        .save_state_snapshot(
-                            snapshot_height,
-                            state_root,
-                            total_supply,
-                            state_data,
-                        )
-                        .await
-                    {
-                        Ok(_) => {
-                            if is_info() {
-                                println!(
-                                    "[INFO][PIPELINE] snapshot_created h={} elapsed_ms={} source=apply_stage",
-                                    snapshot_height,
-                                    start.elapsed().as_millis(),
-                                );
-                            }
-                        }
-                        Err(e) => {
-                            if is_warn() {
-                                println!(
-                                    "[WARN][PIPELINE] snapshot_save_failed h={} err={:?}",
-                                    snapshot_height, e,
-                                );
-                            }
+                    if let Err(e) = storage_for_snapshot.create_state_snapshot(snapshot_height).await {
+                        if is_warn() {
+                            println!("[WARN][PIPELINE] snapshot_create_failed h={} err={:?}", snapshot_height, e);
                         }
                     }
                 });

development/qnet-integration/src/node.rs

@@ -15164,31 +15164,34 @@ impl BlockchainNode {
                     let reg_our_h = crate::unified_p2p::LOCAL_BLOCKCHAIN_HEIGHT.load(std::sync::atomic::Ordering::Relaxed);
                     let reg_best_h = crate::unified_p2p::BEST_PEER_HEIGHT.load(std::sync::atomic::Ordering::Relaxed);
 
+                    // Sync-INDEPENDENT binding-TX rebroadcast: a node's own NodeRegistration must land
+                    // on-chain even while it is still syncing (identity binding is signature-validated,
+                    // not chain-state-validated). This does NOT make an unsynced node VRF-eligible —
+                    // that is gated separately by register_as_active_node_async below. Producer-direct
+                    // gossip + genesis fan-out (same delivery as NodeActivation); re-applying an included
+                    // registration is a nonce/dedup no-op; the attempt budget bounds it.
+                    if let Some(ref p2p) = unified_p2p {
+                        let resend = if let Ok(mut guard) = PENDING_NODE_REGISTRATION.lock() {
+                            let out = if let Some((id, bytes, attempts)) = guard.as_mut() {
+                                if *attempts > 0 { *attempts -= 1; Some((id.clone(), bytes.clone(), *attempts)) } else { None }
+                            } else { None };
+                            if matches!(guard.as_ref(), Some((_, _, 0))) { *guard = None; }
+                            out
+                        } else { None };
+                        if let Some((reg_id, reg_bytes, attempts_left)) = resend {
+                            let _ = p2p.broadcast_transaction(reg_bytes.clone());
+                            let tx_msg = crate::unified_p2p::NetworkMessage::Transaction { data: reg_bytes };
+                            for ip in &crate::unified_p2p::get_genesis_bootstrap_ips() {
+                                p2p.send_network_message(&format!("{}:8001", ip), tx_msg.clone());
+                            }
+                            if is_info() { println!("[INFO][REG] registration_rebroadcast id={} attempts_left={}", reg_id, attempts_left); }
+                        }
+                    }
+
                     if reg_synced {
                         if let Some(ref p2p) = unified_p2p {
                             if is_info() { println!("[INFO][ACTIVE] periodic_registration h={} best={}", reg_our_h, reg_best_h); }
                             p2p.register_as_active_node_async().await;
-
-                            // Bounded rebroadcast of our own NodeRegistration so a dropped join-time
-                            // broadcast still reaches a producer. Re-applying an already-included
-                            // registration is a no-op (nonce/dedup), so this is safe; the budget stops it.
-                            let resend = if let Ok(mut guard) = PENDING_NODE_REGISTRATION.lock() {
-                                let out = if let Some((id, bytes, attempts)) = guard.as_mut() {
-                                    if *attempts > 0 {
-                                        *attempts -= 1;
-                                        Some((id.clone(), bytes.clone(), *attempts))
-                                    } else { None }
-                                } else { None };
-                                if matches!(guard.as_ref(), Some((_, _, 0))) { *guard = None; } // budget spent
-                                out
-                            } else { None };
-                            if let Some((reg_id, reg_bytes, attempts_left)) = resend {
-                                if let Err(e) = p2p.broadcast_transaction(reg_bytes) {
-                                    if is_warn() { println!("[WARN][REG] rebroadcast_fail id={} err={}", reg_id, e); }
-                                } else if is_info() {
-                                    println!("[INFO][REG] registration_rebroadcast id={} attempts_left={}", reg_id, attempts_left);
-                                }
-                            }
                         }
                     } else {
                         if is_info() {
@@ -27914,14 +27917,18 @@ if is_info() { println!("[INFO][SYNC] recovered node={} lag={}", node_id_for_syn
                 if let Ok(mut pend) = PENDING_NODE_REGISTRATION.lock() {
                     *pend = Some((self.node_id.clone(), tx_bytes.clone(), PENDING_REGISTRATION_MAX_REBROADCASTS));
                 }
-                // Broadcast NodeRegistration TX (producer-direct + gossip backup) so a non-producer
-                // node's TX still reaches a producer for inclusion.
+                // Deliver with the same guarantee as NodeActivation: producer-direct gossip + direct
+                // fan-out to every genesis node. A fresh joiner usually has no producer info yet, so a
+                // single gossip is fragile; the genesis fan-out ensures the binding TX reaches the
+                // network even before sync completes.
                 if let Some(ref p2p) = self.unified_p2p {
-                    if let Err(e) = p2p.broadcast_transaction(tx_bytes) {
-                        if is_warn() { println!("[WARN][REG] broadcast_fail hash={}... err={}", &tx_hash[..16.min(tx_hash.len())], e); }
-                    } else {
-                        if is_info() { println!("[INFO][REG] registration_tx_broadcast hash={}", &tx_hash[..16.min(tx_hash.len())]); }
+                    let _ = p2p.broadcast_transaction(tx_bytes.clone());
+                    let tx_msg = crate::unified_p2p::NetworkMessage::Transaction { data: tx_bytes };
+                    let genesis_ips = crate::unified_p2p::get_genesis_bootstrap_ips();
+                    for ip in &genesis_ips {
+                        p2p.send_network_message(&format!("{}:8001", ip), tx_msg.clone());
                     }
+                    if is_info() { println!("[INFO][REG] registration_tx_broadcast hash={} genesis={}", &tx_hash[..16.min(tx_hash.len())], genesis_ips.len()); }
                 }
             } else {
                 eprintln!("[WARN][REG] onchain_tx_failed node={}", self.node_id);

development/qnet-integration/src/rpc.rs

@@ -11352,7 +11352,12 @@ async fn handle_register_node(
                          &wallet_address[..16.min(wallet_address.len())],
                          &tx_hash[..16.min(tx_hash.len())]);
                 if let Some(p2p) = blockchain.get_unified_p2p() {
-                    let _ = p2p.broadcast_transaction(tx_bytes);
+                    let _ = p2p.broadcast_transaction(tx_bytes.clone());
+                    // Same guaranteed delivery as NodeActivation: direct fan-out to all genesis.
+                    let tx_msg = crate::unified_p2p::NetworkMessage::Transaction { data: tx_bytes };
+                    for ip in &crate::unified_p2p::get_genesis_bootstrap_ips() {
+                        p2p.send_network_message(&format!("{}:8001", ip), tx_msg.clone());
+                    }
                 }
             } else {
                 eprintln!("[WARN][REG] super_onchain_tx_failed node={}", node_id);