super-node onboarding: NAT block delivery, keep-up, heartbeat liveness

- transport: reuse the live inbound QUIC conn for NAT/client-dialed peers
  (INBOUND_CONN_BY_LISTEN_ADDR consulted in connect) so genesis shred/serve
  pushes reach a NAT super instead of re-dialing a blocked listen port
- p2p: genesis shred fan-out now includes connected Super peers (was genesis-
  only) so a caught-up super gets per-slot shreds, not a gossip sawtooth
- keep-up: NAT-durable signed-head reply over the live conn; committee-anchored
  head-overclaim clamp (committee + genesis median, scales past 5 nodes)
- heartbeat: dedup on on-chain inclusion (own heartbeat_slots bit) + bounded
  re-anchor, so a missed subwindow is retried not lost -> eligibility holds
- registration: boot fallback gated on synced (single sync-gated source)
- cold-join: reset chain_height on post-promote rehydrate failure
- cleanup: drop dead height_cache + unreachable genesis peer-list branch

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

Author: AIQnetLab

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

@@ -2888,6 +2888,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         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 !qnet_integration::node::coordinator_is_synchronized() {
+            // Defer to the sync-gated activation path (single source of truth). A registration whose
+            // burn-attestors are collected pre-sync (local height ~0 → committee_for_height None →
+            // genesis-set fallback) is rejected post-genesis; sending only once synced binds it to the
+            // true N-2 committee. The sync-gated path re-sends until the registration lands on-chain.
+            if is_info() { println!("[INFO][NODE] activation_send_deferred reason=not_synced path=sync_gated"); }
         } 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/node.rs

@@ -1248,9 +1248,33 @@ pub fn try_advance_finality(round: u64, context: &str) -> bool {
     }
     LAST_FINALIZED_CONSENSUS_ROUND.store(round, std::sync::atomic::Ordering::SeqCst);
     LAST_FINALIZED_HEIGHT.store(round, std::sync::atomic::Ordering::SeqCst);
+    // Release the finality mutex BEFORE the post-advance refresh so the invariant above (no storage I/O
+    // under the lock) holds. On real advance: refresh the committee clamp anchor (once/epoch) + re-sign head.
+    drop(_finality_guard);
+    if let Some(storage) = try_get_storage() {
+        refresh_current_committee(&storage, round);
+    }
+    if let Some(p2p) = try_get_p2p() {
+        p2p.refresh_signed_head_throttled(round);
+    }
     true
 }
 
+/// Refresh the cached committee membership (anti-forgery head-clamp anchor) once per epoch. members =
+/// committee node_ids (empty in the genesis era); genesis_ids always present so the 5 genesis stay an
+/// anchor at any scale. O(R<=1000) recompute under the finality mutex, no-op when the epoch is unchanged.
+fn refresh_current_committee(storage: &Storage, h: u64) {
+    let epoch = h.saturating_sub(1) / 90 + 1;
+    if epoch == crate::unified_p2p::CURRENT_COMMITTEE.read().epoch { return; }
+    let members: std::collections::HashSet<String> = BlockchainNode::committee_for_height(storage, h)
+        .map(|v| v.into_iter().collect())
+        .unwrap_or_default();
+    let genesis_ids: std::collections::HashSet<String> = (1..=crate::genesis_constants::genesis_node_count())
+        .map(|i| format!("genesis_node_{:03}", i)).collect();
+    *crate::unified_p2p::CURRENT_COMMITTEE.write() =
+        std::sync::Arc::new(crate::unified_p2p::CommitteeSnapshot { epoch, members, genesis_ids });
+}
+
 /// v10.2: Atomic height update helper — DISK FIRST, RAM SECOND.
 /// Updates all 3 height sources in one call to prevent inconsistency.
 ///
@@ -10904,8 +10928,13 @@ impl BlockchainNode {
                 println!("[INFO][HEARTBEAT-LOOP] heartbeat emission loop started (spread, unforgeable)");
             }
 
-            // v34: last (epoch, subwindow) for which this node emitted a Heartbeat TX (dedup).
-            let mut last_hb_subwindow: Option<(u64, u64)> = None;
+            // v36: (epoch, subwindow, emit_height) of this node's last Heartbeat emission. Dedup keys on
+            // ON-CHAIN INCLUSION (own heartbeat_slots bit), not emission — a heartbeat that missed
+            // inclusion is re-anchored and retried until the subwindow is recorded, so a transient
+            // desync can never permanently drop a liveness subwindow (producer/reward eligibility).
+            let mut last_hb_emit: Option<(u64, u64, u64)> = None;
+            // Re-anchor cadence: well under HB_ANCHOR_MAX_LAG (90) so a retried heartbeat never goes stale.
+            const HB_REEMIT_INTERVAL: u64 = 60;
 
             while *is_running.read().await {
                 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
@@ -10914,30 +10943,40 @@ impl BlockchainNode {
 
                 // v34: emit ONE unforgeable Heartbeat TX per ~1440-block subwindow (10/epoch) for
                 // super/genesis nodes. Anchored to a recent block hash so it cannot be pre-signed or
-                // backfilled (verified in validate_received_microblock). Replaces self-attested
-                // liveness with on-chain liveness tallied in Account.heartbeat_slots (popcount ≥ 9).
-                // Skip while syncing/behind: a heartbeat anchored to our lagging local height would
-                // be stale at the network tip, and a producer including it would reject the block.
+                // backfilled (verified in validate_received_microblock). Liveness is tallied on-chain
+                // in Account.heartbeat_slots (popcount ≥ 9). Skip while syncing: a heartbeat anchored
+                // to our lagging height would be stale at the tip and the producer would reject it.
                 if !matches!(node_type, NodeType::Light) && current_height >= 2 && !coordinator_is_syncing() {
-                    let hb_epoch = current_height / EMISSION_BLOCK_INTERVAL;
-                    let pos = current_height % EMISSION_BLOCK_INTERVAL;
+                    // Derive epoch/subwindow from the ANCHOR (current_height-2) the heartbeat commits and
+                    // the apply tallies, so the inclusion check tests exactly the bit the emit will set.
+                    let anchor_h = current_height - 2;
+                    let hb_epoch = anchor_h / EMISSION_BLOCK_INTERVAL;
+                    let pos = anchor_h % EMISSION_BLOCK_INTERVAL;
                     let hb_subwindow = pos / 1440;
-                    // Emit at a per-node offset inside the subwindow, not at its boundary, so
-                    // 100k+ nodes spread (~70/block) instead of bursting in one block. Dedup
-                    // makes a late wake (pos already past offset) still emit once in-window.
+                    // Emit at a per-node offset inside the subwindow (not its boundary) so 100k+ nodes
+                    // spread (~70/block) instead of bursting in one block.
                     let hb_offset = Self::heartbeat_offset(&node_id, hb_subwindow);
-                    if last_hb_subwindow != Some((hb_epoch, hb_subwindow)) && (pos % 1440) >= hb_offset {
-                        if let Some(hb_tx) = Self::create_heartbeat_tx_static(&storage, &node_id, current_height).await {
-                            if let Ok(hb_bytes) = bincode::serialize(&hb_tx) {
-                                let hb_gp = hb_tx.gas_price;
-                                if mempool.add_binary_transaction(hb_bytes.clone(), hb_tx.hash.clone(), hb_gp) {
-                                    if let Some(ref p2p) = unified_p2p {
-                                        let _ = p2p.broadcast_transaction(hb_bytes);
-                                    }
-                                    last_hb_subwindow = Some((hb_epoch, hb_subwindow));
-                                    if is_info() {
-                                        println!("[INFO][HEARTBEAT] emitted node={} epoch={} subwindow={} at_h={}",
-                                                 node_id, hb_epoch, hb_subwindow, current_height);
+                    if (pos % 1440) >= hb_offset {
+                        // Done once our own heartbeat for this subwindow is recorded on-chain.
+                        let included = storage.load_account(&node_id).ok().flatten()
+                            .map(|a| a.heartbeat_epoch == hb_epoch
+                                  && (a.heartbeat_slots & (1u16 << hb_subwindow.min(9))) != 0)
+                            .unwrap_or(false);
+                        let first = !matches!(last_hb_emit, Some((e, s, _)) if e == hb_epoch && s == hb_subwindow);
+                        let reanchor = matches!(last_hb_emit, Some((_, _, h)) if current_height.saturating_sub(h) >= HB_REEMIT_INTERVAL);
+                        if !included && (first || reanchor) {
+                            if let Some(hb_tx) = Self::create_heartbeat_tx_static(&storage, &node_id, current_height).await {
+                                if let Ok(hb_bytes) = bincode::serialize(&hb_tx) {
+                                    let hb_gp = hb_tx.gas_price;
+                                    if mempool.add_binary_transaction(hb_bytes.clone(), hb_tx.hash.clone(), hb_gp) {
+                                        if let Some(ref p2p) = unified_p2p {
+                                            let _ = p2p.broadcast_transaction(hb_bytes);
+                                        }
+                                        last_hb_emit = Some((hb_epoch, hb_subwindow, current_height));
+                                        if is_info() {
+                                            println!("[INFO][HEARTBEAT] emitted node={} epoch={} subwindow={} at_h={}",
+                                                     node_id, hb_epoch, hb_subwindow, current_height);
+                                        }
                                     }
                                 }
                             }
@@ -13897,8 +13936,9 @@ impl BlockchainNode {
                                     let mut stall_count = 0u32; // Track consecutive no-progress iterations
 
                                     loop {
-                                        // Re-read current network height (may have advanced)
-                                        let target = crate::unified_p2p::BEST_PEER_HEIGHT.load(Ordering::Relaxed);
+                                        // Re-read current network height (may have advanced) via the clamped
+                                        // chokepoint so a forged head can't inflate the bulk target.
+                                        let target = p2p_clone.get_best_peer_height();
                                         let target = std::cmp::max(target, network_height); // At least initial target
 
                                         // Sync cursor SLAVED to the real applied chain height: always request
@@ -14012,7 +14052,7 @@ impl BlockchainNode {
                                                 // v32.1: re-verify network tip via authoritative peer-quorum height
                                                 // before declaring sync complete. BEST_PEER_HEIGHT alone can lag
                                                 // attestation TTL; pull live quorum max so we don't exit early.
-                                                let best_atomic = crate::unified_p2p::BEST_PEER_HEIGHT.load(Ordering::Relaxed);
+                                                let best_atomic = p2p_clone.get_best_peer_height();
                                                 let quorum_max = p2p_clone.get_max_peer_height();
                                                 let new_target = best_atomic.max(quorum_max);
                                                 if current_from + 3 >= new_target {

development/qnet-integration/src/quic_transport.rs

@@ -261,6 +261,16 @@ static HANDSHAKE_FAIL_TRACKER: once_cell::sync::Lazy<
 static IP_GATE_REJECTS: AtomicU64 = AtomicU64::new(0);
 pub fn ip_gate_reject_count() -> u64 { IP_GATE_REJECTS.load(Ordering::Relaxed) }
 
+/// Live inbound (client-dialed) connections indexed by the peer's advertised QUIC listen addr
+/// (ip : API_PORT + QUIC_PORT_OFFSET). A NAT/client-dialed peer's connection lives under its
+/// EPHEMERAL source addr in the per-transport pool, so a send targeting its listen port misses and
+/// re-dials a port NAT cannot accept inbound. connect() consults this to REUSE the live inbound conn
+/// — the only way to push shreds/blocks to such a peer. Keyed by addr like all existing addressing,
+/// so it inherits (does not worsen) the one-peer-per-advertised-addr assumption.
+static INBOUND_CONN_BY_LISTEN_ADDR: once_cell::sync::Lazy<
+    DashMap<SocketAddr, Arc<QuicConnection>>
+> = once_cell::sync::Lazy::new(DashMap::new);
+
 #[inline]
 fn unix_secs_now() -> u64 {
     std::time::SystemTime::now()
@@ -1270,6 +1280,13 @@ impl QuicTransport {
                     });
 
                     connections_clone.insert(peer_addr, quic_conn.clone());
+                    // Index the inbound conn by the peer's advertised QUIC listen addr so a send that
+                    // targets the listen port (which a NAT/client-dialed peer cannot accept inbound)
+                    // reuses THIS live conn in connect() instead of re-dialing an unreachable port.
+                    INBOUND_CONN_BY_LISTEN_ADDR.insert(
+                        SocketAddr::new(peer_addr.ip(), 8001u16.saturating_add(crate::p2p_transport::QUIC_PORT_OFFSET)),
+                        quic_conn.clone(),
+                    );
                     // v2.96: Promote IP to known tier after successful handshake
                     known_ips_clone.insert(peer_ip_clone, ());
                     if is_info() { println!("[INFO][QUIC] conn_stored peer={} node={} type={} ip_tier=known", get_privacy_id_for_addr(&peer_addr.to_string()), remote_node_id, remote_node_type); }
@@ -1287,7 +1304,14 @@ impl QuicTransport {
                         pid_map.insert(remote_node_id.clone(), peer_api_addr.clone());
                         if is_info() { println!("[INFO][QUIC] peer_mapped node_id={} addr={}", remote_node_id, peer_api_addr); }
                     }
-                    
+
+                    // Register the inbound (client-dialed) peer for signed-head relay reachability:
+                    // is_outbound=false keeps eclipse/reputation/subnet caps; height 0 until its first
+                    // signed HealthPing attests the tip (so it is not quorum-counted early).
+                    if let Some(p2p) = crate::node::try_get_p2p() {
+                        p2p.attest_connected_peer(&remote_node_id, &peer_addr.ip().to_string(), &remote_node_type, 0, false, false);
+                    }
+
                     {
                         let mut s = stats_clone.write().await;
                         s.connections_established += 1;
@@ -1543,6 +1567,10 @@ impl QuicTransport {
         if connections.remove(&peer_addr).is_some() {
             if is_info() { println!("[INFO][QUIC] conn_removed_on_close peer={}", get_privacy_id_for_addr(&peer_addr.to_string())); }
         }
+        // Drop the listen-addr index entry only if it still points at THIS closing conn (a sibling
+        // conn from the same IP must keep its own live entry).
+        let listen_key = SocketAddr::new(peer_addr.ip(), 8001u16.saturating_add(crate::p2p_transport::QUIC_PORT_OFFSET));
+        INBOUND_CONN_BY_LISTEN_ADDR.remove_if(&listen_key, |_, v| Arc::ptr_eq(v, &quic_conn));
     }
 
     /// Handle bidirectional stream (v2.95: length-prefixed protocol for ACK support)
@@ -1644,6 +1672,31 @@ impl QuicTransport {
         let _ = send.finish();
     }
 
+    /// Reply our cached signed head over the LIVE inbound conn (no dial) to a behind pinger — the
+    /// NAT-durable tip feed: it writes on the same accepted QUIC connection the follower's pull uses,
+    /// so it traverses the NAT mapping the follower created (cosend/relay to its listen port cannot).
+    async fn maybe_reply_signed_head(conn: &Arc<QuicConnection>, peer_height: u64) {
+        // The inbound conn already passed the QUIC handshake ML-DSA identity gate and the reply is OUR
+        // OWN signed head, so no per-ping verify here (the follower re-verifies on ingest — I1). Gate only
+        // on the lead: reply iff we are >= HEAD_REPLY_MIN_GAP ahead — suppresses at-tip chatter, O(1)/conn.
+        let head = crate::unified_p2p::LATEST_SIGNED_HEAD.read().clone();
+        let (h_from, h_ts, h_height, h_sig, h_pk) = match head { Some(h) => h, None => return };
+        if h_height < peer_height.saturating_add(crate::unified_p2p::HEAD_REPLY_MIN_GAP) { return; }
+        // Re-emit OUR signed head over THIS live conn (same framing as try_broadcast_once, no dial).
+        let reply = crate::unified_p2p::NetworkMessage::HealthPing { from: h_from, timestamp: h_ts, height: h_height, signature: h_sig, public_key: h_pk };
+        if let Ok(wire) = Self::serialize_message(&reply) {
+            // Bounded like try_broadcast_once so a stalled stream never pins this per-stream task's permit.
+            let _ = tokio::time::timeout(Duration::from_secs(MESSAGE_TIMEOUT_SECS), async {
+                if let Ok(mut send) = conn.connection.open_uni().await {
+                    let len = (wire.len() as u32).to_be_bytes();
+                    let _ = send.write_all(&len).await;
+                    let _ = send.write_all(&wire).await;
+                    let _ = send.finish();
+                }
+            }).await;
+        }
+    }
+
     /// Handle unidirectional stream (broadcast/fire-and-forget messages)
     /// v2.95.2: Updated to read length-prefixed messages (matches try_send_once)
     async fn handle_uni_stream(
@@ -1738,12 +1791,18 @@ impl QuicTransport {
             }
         };
 
+        // Live-conn signed-head reply: serve our cached head back over THIS inbound conn to a behind
+        // pinger (NAT-durable). Capture the pinger height before the handler moves `msg`.
+        let hp_peer_height = if let NetworkMessage::HealthPing { height, .. } = msg { Some(height) } else { None };
         // Call handler
         if let Some(ref h) = handler {
             h(peer_addr, msg);
         }
+        if let Some(ph) = hp_peer_height {
+            Self::maybe_reply_signed_head(&conn, ph).await;
+        }
     }
-    
+
     /// Parse binary message with per-type size enforcement
     fn parse_message(data: &[u8]) -> Result<NetworkMessage, String> {
         if data.len() < 6 {
@@ -1826,6 +1885,16 @@ impl QuicTransport {
             self.connections.remove(&peer_addr);
         }
 
+        // NAT reuse: no outbound conn to this listen addr. Before dialing (which a NAT/client-dialed
+        // peer cannot accept), reuse a live conn the peer opened TO us, indexed by its advertised
+        // listen addr — the only durable channel to push shreds/blocks to such a peer.
+        if let Some(conn) = INBOUND_CONN_BY_LISTEN_ADDR.get(&peer_addr).map(|r| r.value().clone()) {
+            if crate::quic_transport::is_connection_alive(&conn) {
+                return Ok(conn);
+            }
+            INBOUND_CONN_BY_LISTEN_ADDR.remove(&peer_addr);
+        }
+
         // v2.96: Per-peer reconnect cooldown — reject if last attempt was < COOLDOWN ago.
         // This is the primary defense against reconnect storms: 5 subsystems calling connect()
         // simultaneously all get instant rejection except the first one within the window.
@@ -1999,7 +2068,7 @@ impl QuicTransport {
             if remote_node_id != self.node_id {
                 if let Some(p2p) = crate::node::try_get_p2p() {
                     let ip = connection.remote_address().ip().to_string();
-                    p2p.attest_connected_peer(&remote_node_id, &ip, &remote_node_type, remote_block_height, remote_verified);
+                    p2p.attest_connected_peer(&remote_node_id, &ip, &remote_node_type, remote_block_height, remote_verified, true);
                 }
             }
         }