pipeline: close the verify-to-commit park race, make the deferred TTL unconditional

At load the network wedged at h~1350 on four of five nodes, each at a slightly
different height. Measured chain: a child block checks for its parent via
canonical_hash_at, which reads storage — and storage only answers at APPLY-commit,
600-850 ms after the parent VERIFIES under ~250-tx blocks. Children of a verified
parent are released at verify time. A child arriving inside that verify-to-commit
window finds neither: the storage read misses, the release has already run, and the
block parks forever. The re-download of parked heights used to hide this by
accident; the byte-budget sync's deferred-height mask removed exactly that, so one
parked orphan turned into a permanent per-node wedge ("delivered rounds=0" loop).

Fix: a task-local verified_recent map (hash -> height), inserted where the release
runs, consulted as a guard before parking (`Ok(None) if parent_verified_in_flight`).
The guard, the insert and the release all execute in the SAME pipeline task, so the
ordering is total: a parent either verified earlier and the guard sees it, or
verifies later and its release drains the parked child. No locks, no timing window,
no storage round-trip. Semantics match the existing release exactly — any VERIFIED
parent frees its children; apply still arbitrates the canonical chain. Entries are
pruned 500 below the tip past 1024.

A first attempt re-checked storage after parking; it was rejected in review because
the re-check lands inside the same verify-to-commit window it is trying to close.

The deferred TTL sweep now runs whenever the buffer is non-empty. It was gated
behind count > 100, so a lone orphan below the gate lived forever — and with the
deferred set masking its height from sync, that single entry wedged the node.

Tests: 420 green, zero warnings.

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

Author: AIQnetLab

development/qnet-integration/src/block_pipeline.rs

@@ -1812,6 +1812,11 @@ impl BlockPipeline {
         let mut deferred_count: usize = 0;
         // Byte occupancy: count-only bounds let 2000 multi-MB blocks pin gigabytes of RAM.
         let mut deferred_bytes: usize = 0;
+        // Verified-but-not-yet-applied parent hashes. Storage only answers at apply-commit
+        // (600+ ms later under load) while children release at VERIFY — a child arriving in
+        // that window found neither and parked forever. Same task as the release, so reads
+        // here are exact, with no lock and no timing.
+        let mut verified_recent: HashMap<[u8; 32], u64> = HashMap::new();
         // A restarted pipeline dropped its buffer; stale holds would mask sync forever.
         DEFERRED_HEIGHTS.clear();
         // Per-producer occupancy, maintained incrementally: counting by scanning the whole buffer
@@ -1975,8 +1980,13 @@ impl BlockPipeline {
                         );
                     }
                 }
+                let parent_verified_in_flight =
+                    verified_recent.get(&mb.previous_hash).copied() == Some(mb.height.saturating_sub(1));
                 let prev_hash_ok = match load_result {
                     Ok(Some(prev_hash)) => mb.previous_hash == prev_hash,
+                    // Parent verified in this loop but its apply-commit hasn't reached storage
+                    // yet — proceed; FIFO to the apply stage preserves parent-before-child.
+                    Ok(None) if parent_verified_in_flight => true,
                     Ok(None) => {
                         // Capture height fields BEFORE moving `decoded` into
                         // the deferred map — `mb` is borrowed from `decoded`
@@ -2068,9 +2078,13 @@ impl BlockPipeline {
                             let parked_at = std::time::SystemTime::now()
                                 .duration_since(std::time::UNIX_EPOCH)
                                 .map(|d| d.as_secs()).unwrap_or(0);
+                            let parked_hash = decoded.microblock.hash();
                             let waiters = deferred.entry(mb.previous_hash).or_default();
                             // Drop an exact duplicate re-delivery; distinct siblings both survive.
-                            if !waiters.iter().any(|(_, d)| d.microblock.hash() == decoded.microblock.hash()) {
+                            // No recheck needed after this insert: the verified-in-flight guard above
+                            // runs in the SAME task as the release, so a parent is either seen there
+                            // or verifies later and drains us.
+                            if !waiters.iter().any(|(_, d)| d.microblock.hash() == parked_hash) {
                                 *deferred_by_producer.entry(decoded.microblock.producer.clone()).or_insert(0) += 1;
                                 deferred_track(child_h);
                                 deferred_bytes = deferred_bytes.saturating_add(incoming_sz);
@@ -3065,6 +3079,8 @@ impl BlockPipeline {
             let block_height = decoded.height;
             // Identity of the block just verified — the key its waiting children were parked under.
             let verified_hash = decoded.microblock.hash();
+            // Answers "parent verified, apply-commit pending" for the parking guard above.
+            verified_recent.insert(verified_hash, block_height);
 
             // Liveness is NOT recorded here. A signature-verified block only proves the producer
             // signed something — a block that fails apply (bad state_root, unresolvable pk, breaker)
@@ -3132,7 +3148,17 @@ impl BlockPipeline {
             // enough: blocks parked ABOVE the tip are never "behind" it, and during the very stall
             // where this buffer matters the tip does not advance — so a height-only rule can never
             // reclaim them. An age rule always can.
-            if deferred_count > 100 {
+            // Verified-in-flight entries are only useful until apply-commit lands in storage;
+            // anything 500 below the tip is long-committed (or long-dead).
+            if verified_recent.len() > 1024 {
+                let tip = crate::unified_p2p::LOCAL_BLOCKCHAIN_HEIGHT.load(Ordering::Relaxed);
+                verified_recent.retain(|_, h| h.saturating_add(500) > tip);
+            }
+
+            // Unconditional: the TTL is an invariant, not a large-buffer-only rule. A single
+            // orphan parked below the old >100 gate lived forever — and with the deferred set
+            // masking its height from sync, that single entry wedged the whole node.
+            if deferred_count > 0 {
                 let chain_h = storage.get_chain_height().unwrap_or(0);
                 {
                     const DEFERRED_MAX_AGE_SECS: u64 = 120;