confirm: probe in-flight outs once, not per parent

CHANGELOG.md

@@ -54,6 +54,13 @@ before 1.0).
 
 ### Changed
 
+- **Load in-flight outs: iterate pins, not parents×layers.** Post-milestone
+  unique spent parents (~160k/window) were probing every in-flight layer
+  via `get_out` in `fill_missing_parent_ranges` and pin adopt. Union
+  `out_ids` / `for_each_out` instead. `ibd: perf` `stamp_sub` prints
+  **`fill=`** / **`idx=`** for that wall (it was inside `batch=` with no
+  token).
+
 - **Pin `merge_outs` no-op Arc:** empty / already-covered `checked`+`live`
   keep the outs Arc (assemble sticky `ptr_eq`). RCU compose borrows `live`
   instead of cloning script bytes on every retry.

OPERATOR.md

@@ -211,7 +211,7 @@ Requires **tip mode** (`node: catch-up complete … tip tracking`). During IBD u
 | Line | Level | Use |
 |------|-------|-----|
 | `ibd: progress` | INFO | Tip rate, `loadq`/`scriptq`/`writeq`, `txs=` (Class A / `tx.idx` count), horizon, tip ETA, **`bq soft=n/win RAM=`** (in-RAM body queue; soft densify: under ~100 MiB free ahead, over that only ~1 min confirm window, at/over 1 GiB assign-stop fill holes in the fetched range only) |
-| `ibd: perf` | DEBUG | Inflight + **`bq soft= RAM=`**; **`load=`** is pin+assemble only. **`load_thr pack/stamp/pin/asm/prune`** is the load OS thread. **`stamp=`** nests **`pack=`** (plan HashMap) vs **`head=`** (leftover TipOnly). After a published wave `head=` is ~0. **`script=`** is verify ns (`jobs=` / `skip=`); recv/send are wait. **`lookup_thr keep=`** is live-union splice. **`pin_txid=`** vs leftover `tx.head` |
+| `ibd: perf` | DEBUG | Inflight + **`bq soft= RAM=`**; **`load=`** is pin+assemble only. **`load_thr pack/stamp/pin/asm/prune`** is the load OS thread. **`stamp=`** nests **`pack=`** (plan HashMap) vs **`head=`** (leftover TipOnly) vs **`fill=`** (`fill_missing`; **`idx=`** spent/body ranges). After a published wave `head=` is ~0. **`script=`** is verify ns (`jobs=` / `skip=`); recv/send are wait. **`lookup_thr keep=`** is live-union splice. **`pin_txid=`** vs leftover `tx.head` |
 | `ibd: sizes` | DEBUG | RSS + work path + **`bq soft=` / `RAM=`** + **conf_plans** + confirm pipe |
 | `ibd: perf_dbg` | DEBUG | µs/blk load/write, pin/edge detail, **plan_batch** (`us/pin_txid` vs `probe/idx/body us/key`) + **class_a commit** |
 

crates/rbitcoin-consensus/src/confirm_run/pin.rs

@@ -129,15 +129,12 @@ pub(super) fn pin_for_wire_batch(
     }
 
     if let Some(ifo) = in_flight {
-        for (id, need) in &parent_vouts {
-            if plan_by_id.contains_key(id) {
-                continue;
+        ifo.for_each_out(|id, pin| {
+            if plan_by_id.contains_key(&id) || !parent_vouts.contains_key(&id) {
+                return;
             }
-            if let Some(pin) = ifo.get_out(*id) {
-                let _ = need;
-                plan_by_id.insert(*id, std::sync::Arc::clone(pin));
-            }
-        }
+            plan_by_id.insert(id, std::sync::Arc::clone(pin));
+        });
     }
     for (id, _need) in &parent_vouts {
         if plan_by_id.contains_key(id) {

crates/rbitcoin-net/src/ibd/perf_log.rs

@@ -41,7 +41,8 @@
 //! `scriptq` can stay empty. Prefer `lookup_thr busy=` / `thr load=busy/wait=` /
 //! `ready=` + `scriptq_hwm=` (OS-thread occupancy + queue high-water). High
 //! `load_thr stamp=` nests `pack=` (plan HashMap) vs `head=` (leftover TipOnly
-//! `prep_head_fk_ns`). After a published wave `head=` is ~0. High `head=` +
+//! `prep_head_fk_ns`) vs `fill=` (`fill_missing_parent_ranges`; `idx=` is
+//! spent/body range batches). After a published wave `head=` is ~0. High `head=` +
 //! `ready>0` + `scriptq=1` ⇒ leftover TipOnly on load, not “scripts hungry.”
 //! High load_recv_wait + ready=0 ⇒ lookup is the pole.
 //!
@@ -346,6 +347,9 @@ pub(crate) struct IbdPerfSample {
     pub stamp_batch_head_fk_ms: u64,
     pub stamp_batch_stamp_ms: u64,
     pub stamp_batch_finish_ms: u64,
+    /// `fill_missing_parent_ranges` (skip walk + idx). Nested `idx=` is spent/body range batches.
+    pub stamp_batch_fill_ms: u64,
+    pub stamp_batch_fill_idx_ms: u64,
     pub thr_load_recv_wait_ms: u64,
     pub thr_load_pack_ms: u64,
     pub thr_load_clone_ms: u64,
@@ -436,6 +440,8 @@ pub(crate) struct IbdPerfSample {
     pub arch_prep_body_lookups: u64,
     pub arch_prep_stamp_ms: u64,
     pub arch_prep_finish_ms: u64,
+    pub arch_prep_fill_ms: u64,
+    pub arch_prep_fill_idx_ms: u64,
     pub arch_write_total_ms: u64,
     pub arch_write_reserve_ms: u64,
     pub arch_write_body_ms: u64,
@@ -609,6 +615,8 @@ impl Default for IbdPerfSample {
             stamp_batch_head_fk_ms: 0,
             stamp_batch_stamp_ms: 0,
             stamp_batch_finish_ms: 0,
+            stamp_batch_fill_ms: 0,
+            stamp_batch_fill_idx_ms: 0,
             thr_load_recv_wait_ms: 0,
             thr_load_pack_ms: 0,
             thr_load_clone_ms: 0,
@@ -686,6 +694,8 @@ impl Default for IbdPerfSample {
             arch_prep_body_lookups: 0,
             arch_prep_stamp_ms: 0,
             arch_prep_finish_ms: 0,
+            arch_prep_fill_ms: 0,
+            arch_prep_fill_idx_ms: 0,
             arch_write_total_ms: 0,
             arch_write_reserve_ms: 0,
             arch_write_body_ms: 0,
@@ -1072,6 +1082,8 @@ pub(crate) fn sample(
         stamp_batch_head_fk_ms: ns_ms(arch_res.prep_head_fk_ns),
         stamp_batch_stamp_ms: ns_ms(arch_res.prep_stamp_ns),
         stamp_batch_finish_ms: ns_ms(arch_res.prep_finish_ns),
+        stamp_batch_fill_ms: ns_ms(arch_res.prep_fill_ns),
+        stamp_batch_fill_idx_ms: ns_ms(arch_res.prep_fill_idx_ns),
         thr_load_recv_wait_ms: ns_ms(thr.load_recv_wait_ns),
         thr_load_pack_ms: ns_ms(thr.load_pack_ns),
         thr_load_clone_ms: ns_ms(thr.load_clone_ns),
@@ -1149,6 +1161,8 @@ pub(crate) fn sample(
         arch_prep_body_lookups: head_res.body_lookups,
         arch_prep_stamp_ms: ns_ms(arch_res.prep_stamp_ns),
         arch_prep_finish_ms: ns_ms(arch_res.prep_finish_ns),
+        arch_prep_fill_ms: ns_ms(arch_res.prep_fill_ns),
+        arch_prep_fill_idx_ms: ns_ms(arch_res.prep_fill_idx_ns),
         arch_write_total_ms: ns_ms(arch_res.write_total_ns),
         arch_write_reserve_ms: ns_ms(arch_res.write_reserve_ns),
         arch_write_body_ms: ns_ms(arch_res.write_body_ns),
@@ -1209,6 +1223,7 @@ fn plan_batch_ms(s: &IbdPerfSample) -> u64 {
         .saturating_add(s.arch_prep_inflight_ms)
         .saturating_add(s.arch_prep_head_ms)
         .saturating_add(s.arch_prep_stamp_ms)
+        .saturating_add(s.arch_prep_fill_ms)
         .saturating_add(s.arch_prep_finish_ms)
 }
 
@@ -1317,7 +1332,7 @@ pub(crate) fn format_info(s: &IbdPerfSample) -> String {
              batch_assign={}ms collect={}ms pin_txid={} pin_txid%={} pin_txid_ms={} \
              leftover_n={} leftover_hit={} leftover_ms={} leftover_pend={} leftover_cdf0={} leftover_cdf3={} leftover_age_n={} \
              recent={} recent_ms={} \
-             head={}ms stamp={}ms finish={}ms)",
+             head={}ms stamp={}ms fill={}ms(idx={}ms) finish={}ms)",
             s.stamp_struct_ms,
             s.stamp_struct_txid_ms,
             s.stamp_struct_walk_ms,
@@ -1340,6 +1355,8 @@ pub(crate) fn format_info(s: &IbdPerfSample) -> String {
             s.arch_recent_ms,
             s.stamp_batch_head_ms,
             s.stamp_batch_stamp_ms,
+            s.stamp_batch_fill_ms,
+            s.stamp_batch_fill_idx_ms,
             s.stamp_batch_finish_ms,
         ));
     }
@@ -1701,7 +1718,7 @@ pub(crate) fn format_debug(s: &IbdPerfSample) -> String {
         out.push_str(&format!(
             " | plan_batch assign={} collect={} inflight={} pin_txid={}/{} pin_txid_ms={} \
              us/pin_txid={} recent={} recent_ms={} head_fk={} head={} \
-             stamp={} finish={} resolve_us/blk={} ext={} head_hit={}/{} \
+             stamp={} finish={} fill={} fill_idx={} resolve_us/blk={} ext={} head_hit={}/{} \
              stamp_n batch={}",
             s.arch_prep_assign_ms,
             s.arch_prep_collect_ms,
@@ -1716,6 +1733,8 @@ pub(crate) fn format_debug(s: &IbdPerfSample) -> String {
             s.arch_prep_head_ms,
             s.arch_prep_stamp_ms,
             s.arch_prep_finish_ms,
+            s.arch_prep_fill_ms,
+            s.arch_prep_fill_idx_ms,
             resolve_us_blk,
             s.arch_ext_need,
             s.arch_head_hit,
@@ -2380,6 +2399,8 @@ mod tests {
         s.stamp_batch_head_ms = 2;
         s.stamp_batch_stamp_ms = 1;
         s.stamp_batch_finish_ms = 1;
+        s.stamp_batch_fill_ms = 8;
+        s.stamp_batch_fill_idx_ms = 7;
         s.arch_prep_age_hit_n = 50;
         s.arch_prep_age_cdf0_pct = 10;
         s.arch_prep_age_cdf3_pct = 40;
@@ -2394,6 +2415,8 @@ mod tests {
         s.arch_prep_head_ms = 3;
         s.arch_prep_stamp_ms = 1;
         s.arch_prep_finish_ms = 1;
+        s.arch_prep_fill_ms = 8;
+        s.arch_prep_fill_idx_ms = 7;
         s.arch_resolve_ns = 8_000_000;
         s.arch_resolve_blocks = 4;
         s.arch_head_hit = 20;
@@ -2430,6 +2453,7 @@ mod tests {
         s.stamp_struct_walk_ms = 2;
         let info = format_info(&s);
         assert!(info.contains("stamp_sub("), "{info}");
+        assert!(info.contains("fill=8ms(idx=7ms)"), "{info}");
         assert!(info.contains("struct_txid=6ms"), "{info}");
         assert!(info.contains("struct_walk=2ms"), "{info}");
         assert!(info.contains("pin_txid=15"), "{info}");

crates/rbitcoin-query/src/archive.rs

@@ -1852,6 +1852,40 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    /// In-flight CreatePin on the oldest layer still skips idx (no store row).
+    #[test]
+    fn fill_missing_skips_idx_when_oldest_layer_has_outs() {
+        let (dir, q) = temp_query("fill-skip-oldest-outs");
+        let pin = std::sync::Arc::new((
+            TxRecord {
+                txid: [1u8; 32],
+                version: 1,
+                locktime: 0,
+                input_start_fk: Fk::NULL,
+                input_count: 1,
+                output_start_fk: Fk::NULL,
+                output_count: 1,
+            },
+            vec![OutputRecord::unspent(1, vec![0x51])],
+        ));
+        let mut log = crate::InFlightLog::new();
+        log.note_layer(crate::InFlightLayer::from_plan_pins([(Fk(1), &pin)]));
+        for i in 10u8..18 {
+            let mut tid = [0u8; 32];
+            tid[0] = i;
+            log.note_layer(crate::InFlightLayer::from_txid_fks([(tid, Fk(i as u64))]));
+        }
+        let mut idents = crate::U64Map::default();
+        idents.insert(1, crate::ParentIdent::new([1u8; 32]));
+        crate::fill_missing_parent_ranges(q.store(), &log.snapshot(), &mut idents)
+            .expect("in-flight outs skip idx even on the oldest layer");
+        assert!(
+            idents.get(&1).and_then(|p| p.body).is_none(),
+            "skip must not invent a body_range"
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn archive_filter_need_header_fks_drops_archived() {
         use rbitcoin_store::HeaderRecord;

crates/rbitcoin-query/src/in_flight.rs

@@ -12,8 +12,9 @@
 //! layer (Class C ∥ drain/seal). Call after pin — stamp skips `body_range`
 //! when this map still has CreatePin outs.
 //!
-//! Lookup is newest→oldest scan over layers (O(L)); pack counts are small and
-//! L is bounded by pipeline queue depth.
+//! Per-key [`InFlightView::get_out`] / `get_create_fk` is newest→oldest (O(L)).
+//! Bulk skip/adopt over many parents must iterate outs (O(pins)), not
+//! parents×L — post-milestone unique parents dwarf pipeline CreatePins.
 
 use crate::archive::CreatePin;
 use crate::U64Map;
@@ -255,6 +256,30 @@ impl InFlightView {
         None
     }
 
+    /// Union of CreatePin keys across layers.
+    pub fn out_ids(&self) -> crate::U64Set {
+        let n: usize = self.layers.iter().map(|l| l.outs.len()).sum();
+        let mut ids = crate::U64Set::with_capacity_and_hasher(n, Default::default());
+        for layer in self.layers.iter() {
+            for id in layer.outs.keys() {
+                ids.insert(*id);
+            }
+        }
+        ids
+    }
+
+    /// Newest layer first; each id once (same winner as [`Self::get_out`]).
+    pub fn for_each_out(&self, mut f: impl FnMut(u64, &CreatePin)) {
+        let mut seen = crate::U64Set::default();
+        for layer in self.layers.iter().rev() {
+            for (id, pin) in &layer.outs {
+                if seen.insert(*id) {
+                    f(*id, pin);
+                }
+            }
+        }
+    }
+
     /// Create fk for txid from prior uncommitted packs.
     #[inline]
     pub fn get_create_fk(&self, txid: &[u8; 32]) -> Option<Fk> {
@@ -327,6 +352,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn out_ids_unions_layers_and_matches_get_out() {
+        let mut log = InFlightLog::new();
+        let a = pin(1);
+        let b = pin(2);
+        log.note_layer(InFlightLayer::from_plan_pins([(Fk(1), &a)]));
+        log.note_layer(InFlightLayer::from_plan_pins([(Fk(2), &b)]));
+        let v = log.snapshot();
+        let ids = v.out_ids();
+        assert!(ids.contains(&1));
+        assert!(ids.contains(&2));
+        assert_eq!(ids.len(), 2);
+        assert!(v.get_out(1).is_some());
+        assert!(v.get_out(99).is_none());
+        assert!(!ids.contains(&99));
+        let mut newest = None;
+        v.for_each_out(|id, _| {
+            if newest.is_none() {
+                newest = Some(id);
+            }
+        });
+        assert_eq!(newest, Some(2), "newest layer first");
+    }
+
     #[test]
     fn note_does_not_mutate_prior_layer_arcs() {
         let mut log = InFlightLog::new();

crates/rbitcoin-query/src/lib.rs

@@ -474,6 +474,10 @@ pub mod archive_phase_stats {
     pub static RESOLVED_STAMP: AtomicU64 = AtomicU64::new(0);
     /// `fill_missing_parent_ranges` entries (stamp + optional prestamp).
     pub static FILL_MISSING_N: AtomicU64 = AtomicU64::new(0);
+    /// Wall of `fill_missing_parent_ranges` (skip walk + idx).
+    pub static FILL_MISSING_NS: AtomicU64 = AtomicU64::new(0);
+    /// `tx_body_range_batch` + `tx_spent_range_batch` inside fill_missing.
+    pub static FILL_MISSING_IDX_NS: AtomicU64 = AtomicU64::new(0);
 
     /// Full load batch wall (struct → lookup → enqueue wait).
     pub static PREP_TOTAL_NS: AtomicU64 = AtomicU64::new(0);
@@ -534,6 +538,8 @@ pub mod archive_phase_stats {
         pub prep_head_fk_ns: u64,
         pub prep_stamp_ns: u64,
         pub prep_finish_ns: u64,
+        pub prep_fill_ns: u64,
+        pub prep_fill_idx_ns: u64,
         pub prep_publish_ns: u64,
         pub prep_qwait_ns: u64,
         pub prep_blocks: u64,
@@ -558,6 +564,7 @@ pub mod archive_phase_stats {
                 .saturating_add(self.prep_head_ns)
                 .saturating_add(self.prep_stamp_ns)
                 .saturating_add(self.prep_finish_ns)
+                .saturating_add(self.prep_fill_ns)
                 .saturating_add(self.prep_publish_ns)
                 .saturating_add(self.prep_qwait_ns)
         }
@@ -623,6 +630,8 @@ pub mod archive_phase_stats {
             prep_head_fk_ns: prep_head_fk,
             prep_stamp_ns: PREP_STAMP_NS.swap(0, Ordering::Relaxed),
             prep_finish_ns: PREP_FINISH_NS.swap(0, Ordering::Relaxed),
+            prep_fill_ns: FILL_MISSING_NS.swap(0, Ordering::Relaxed),
+            prep_fill_idx_ns: FILL_MISSING_IDX_NS.swap(0, Ordering::Relaxed),
             prep_publish_ns: PREP_PUBLISH_NS.swap(0, Ordering::Relaxed),
             prep_qwait_ns: PREP_QWAIT_NS.swap(0, Ordering::Relaxed),
             prep_blocks: PREP_BLOCKS.swap(0, Ordering::Relaxed),
@@ -677,6 +686,14 @@ pub mod archive_phase_stats {
         });
     }
 
+    #[inline]
+    pub fn note_fill_missing_ns(all_ns: u64, idx_ns: u64) {
+        exclusive::with(|| {
+            add(&FILL_MISSING_NS, all_ns);
+            add(&FILL_MISSING_IDX_NS, idx_ns);
+        });
+    }
+
     /// Leftover TipOnly pending hits + winner age buckets (load stamp).
     #[inline]
     pub fn note_leftover_mix(pend: u64, age0: u64, age3: u64, age_n: u64) {

crates/rbitcoin-query/src/stamp.rs

@@ -215,10 +215,12 @@ pub fn fill_missing_parent_ranges(
     idents: &mut U64Map<ParentIdent>,
 ) -> Result<(), QueryError> {
     crate::archive_phase_stats::note_fill_missing();
+    let t_all = Instant::now();
+    let inflight_outs = in_flight.out_ids();
     let mut need_body: Vec<Fk> = Vec::new();
     let mut need_spent: Vec<Fk> = Vec::new();
     for (&id, ident) in idents.iter() {
-        if in_flight.get_out(id).is_some() {
+        if inflight_outs.contains(&id) {
             continue;
         }
         let fk = Fk(id);
@@ -229,6 +231,7 @@ pub fn fill_missing_parent_ranges(
             need_spent.push(fk);
         }
     }
+    let t_idx = Instant::now();
     let mut body_filled = U64Set::default();
     if !need_body.is_empty() {
         let filled = store.tx_body_range_batch(&need_body)?;
@@ -270,5 +273,7 @@ pub fn fill_missing_parent_ranges(
             }
         }
     }
+    let idx_ns = t_idx.elapsed().as_nanos() as u64;
+    crate::archive_phase_stats::note_fill_missing_ns(t_all.elapsed().as_nanos() as u64, idx_ns);
     Ok(())
 }