query: bound write loc packs to overlapping lookup write

CHANGELOG.md

@@ -108,10 +108,13 @@ before 1.0).
   ENOENT / `.tmp` so a live seal worker unlinking the OA cannot fail the
   snapshot. Live table drops before cleanup.
 
-- **Write loc RAM:** Class A append returns loc pairs; write keeps them in a
-  sequential window and stamps same-batch / just-written abs from that RAM
-  (packed pin outs). Write does not pread `create.loc`. Missing stamp is
-  `Corrupt`.
+- **Write loc RAM:** Class A append returns loc pairs; write keeps them in
+  height-tagged **thread-local** packs (no `Query` mutex) until write of
+  `lookup_started_hi` at note, extended while the pack is still at/above the
+  load drain fence (InFlight still makes TipOnly skip disk loc). Same-write
+  prune keeps the noting pack. Same-batch / just-written abs stamp from that
+  RAM (packed pin outs). Disconnect is polled on the write thread. Write does
+  not pread `create.loc`. Missing stamp is `Corrupt`. `ibd: sizes` `wloc=`.
 
 - **`create.loc` leftover stamp:** lookup reads/sums only through the highest
   fk in each 1024-create window, preads those windows as one bulk batch (held

SCHEMA.md

@@ -377,8 +377,9 @@ head-resolve session, else `pread_batch`). Non-overflow windows use a SIMD
 prefix sum (`u8×8` SSE2 on x86_64, NEON on aarch64).
 
 One `create_loc_range_batch` yields both `(txout, spent)` and `n_out`. Lookup
-stamps both ranges; load copies the stamp; write appends loc and keeps the RAM
-pairs (same-batch abs). Write does not pread `create.loc`. Occupied 21 Class A
+stamps both ranges; load copies the stamp; write appends loc and keeps RAM
+packs until write of `lookup_started_hi` at note (just-written abs). Write
+does not pread `create.loc`. Occupied 21 Class A
 is refused. Leftover `{txout,spent,inwit}.idx` and `spent.off` are unlinked on
 empty 21/22 open.
 

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

@@ -64,6 +64,9 @@ pub fn confirm_write_phase(
                 .map(|p| (p.height.0, p.hash))
                 .collect();
             finish_post_commit_hashes(query, &items)?;
+            if let Some(h) = items.iter().map(|(h, _)| *h).max() {
+                query.prune_write_create_loc(h);
+            }
             return Ok(Vec::new());
         }
         WriteBatchVsTip::SpansTip => {
@@ -109,7 +112,8 @@ pub fn confirm_write_phase(
             // must already have lookup stamps — missing abs is Corrupt.
             // Direct SH collect is a no-op — skip the FkMap.
             if committed {
-                query.note_write_create_loc(&planned_fks, &loc);
+                let pack_hi = batch.prepared.last().map(|p| p.height.0).unwrap_or(0);
+                query.note_write_create_loc(&planned_fks, &loc, pack_hi);
                 if query.index_mode().is_tip() {
                     let t_map = Instant::now();
                     write_create_pins.reserve(planned_fks.len());
@@ -239,6 +243,9 @@ pub fn confirm_write_phase(
     }
 
     // No tip GC of sparse pins (dropped with ScriptOkBatch).
+    if let Some(h) = batch.prepared.iter().map(|p| p.height.0).max() {
+        query.prune_write_create_loc(h);
+    }
     rbitcoin_query::note_confirm(&query.confirm_stats().phase_blocks, n_blocks as u64);
     query
         .confirm_stats()
@@ -398,7 +405,8 @@ fn annotate_jobs_from_connected_hash(
 }
 
 /// After Class A commit, stamp spend creates from append RAM loc
-/// (this pack + just-written packs). Write never preads `create.loc`.
+/// (this pack + just-written packs still in the write loc window).
+/// Write never preads `create.loc`.
 pub(super) fn fill_planned_create_layout_after_commit(
     query: &Query,
     batch_parents: &mut rbitcoin_query::BatchParents,

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

@@ -1259,7 +1259,7 @@ fn fill_same_batch_abs_from_append_loc_ram() {
         &mut bp,
         &fks,
         &loc,
-        &[parent_pin, child_pin],
+        &[std::sync::Arc::clone(&parent_pin), child_pin],
         &prepared,
     )
     .expect("same-batch fill from append RAM");
@@ -1275,6 +1275,51 @@ fn fill_same_batch_abs_from_append_loc_ram() {
         "write fill/ensure must not pread create.loc"
     );
 
+    q.set_lookup_started_hi(Some(4));
+    q.note_write_create_loc(&fks, &loc, 1);
+    let mut bp_later = BatchParents::new();
+    bp_later.insert_create_pin(
+        fks[0],
+        std::sync::Arc::clone(&parent_pin),
+        vec![0],
+        None,
+        None,
+        Vec::new(),
+    );
+    assert!(!bp_later.has_abs_layout(fks[0]));
+    let later = [Prepared {
+        height: Height(2),
+        header_fk: Fk(2),
+        tx_fks: vec![Fk(3)],
+        jobs: vec![],
+        spends: vec![([0x32u8; 32], 0, Fk(3), fks[0], 0)],
+        fees: 0,
+        check_scripts: false,
+        time: 1,
+        bits: bitcoin::CompactTarget::from_consensus(0x207f_ffff),
+        hash: [8u8; 32],
+        txids: vec![],
+        prev_mtp: 0,
+    }];
+    q.store().reset_spent_range_batch();
+    fill_planned_create_layout_after_commit(&q, &mut bp_later, &[], &[], &[], &later)
+        .expect("just-written fill from write loc RAM");
+    assert!(bp_later.has_abs_layout(fks[0]));
+    assert_eq!(
+        bp_later.get_spender_abs(fks[0], 0),
+        Some(rbitcoin_store::spent_abs(loc[0].spent.0, 0))
+    );
+    q.prune_write_create_loc(3);
+    assert!(
+        q.write_create_loc(fks[0]).is_some(),
+        "keep until write of lookup_started_hi (and pack below drain fence)"
+    );
+    q.prune_write_create_loc(4);
+    assert!(
+        q.write_create_loc(fks[0]).is_none(),
+        "drop after last overlapping lookup batch finished write"
+    );
+
     let _ = std::fs::remove_dir_all(&path);
 }
 

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

@@ -1663,6 +1663,7 @@ pub(crate) fn format_sizes(s: &IbdPerfSample) -> String {
     let file_pct = div_or_0(100 * s.rss_file_kb, s.rss_kb);
     let bq_mib = s.bq_bytes / (1024 * 1024);
     let if_mib = o.inflight_bytes / (1024 * 1024);
+    let wloc_mib = o.wloc_bytes / (1024 * 1024);
     let h2h_mib = (o.h2h_keys as u64).saturating_mul(48) / (1024 * 1024);
     let fence_mib = (o.fence_runs as u64).saturating_mul(16) / (1024 * 1024);
     let conf_wire_mib = (load_wire_mib
@@ -1674,6 +1675,7 @@ pub(crate) fn format_sizes(s: &IbdPerfSample) -> String {
     let class_c_l2_mib = h.class_c_l2_bytes / (1024 * 1024);
     let accounted_mib = bq_mib
         .saturating_add(if_mib)
+        .saturating_add(wloc_mib)
         .saturating_add(h2h_mib)
         .saturating_add(fence_mib)
         .saturating_add(conf_wire_mib)
@@ -1691,7 +1693,7 @@ pub(crate) fn format_sizes(s: &IbdPerfSample) -> String {
          | conf_plans={} \
          | conf loadq={}/{} blks={} wire={}MiB scriptq={}/{} blks={} wire={}MiB writeq={}/{} blks={} wire={}MiB parents={} \
            feed ready={} inflight={} \
-         | heap bq={}MiB iflight={}L/{}pin≈{}MiB \
+         | heap bq={}MiB iflight={}L/{}pin≈{}MiB wloc={}L/{}pair≈{}MiB \
            h2h={}k≈{}MiB fence={}≈{}MiB \
            wire={}MiB fuse8={}MiB mphf_g={}MiB open_keys={}MiB class_c_l2={}MiB \
            accounted≈{}MiB residual≈{}MiB \
@@ -1739,6 +1741,9 @@ pub(crate) fn format_sizes(s: &IbdPerfSample) -> String {
         o.inflight_layers,
         o.inflight_pins,
         if_mib,
+        o.wloc_packs,
+        o.wloc_pairs,
+        wloc_mib,
         o.h2h_keys,
         h2h_mib,
         o.fence_runs,
@@ -2425,6 +2430,9 @@ mod tests {
         s.owned.inflight_layers = 3;
         s.owned.inflight_pins = 12_000;
         s.owned.inflight_bytes = 48 * 1024 * 1024;
+        s.owned.wloc_packs = 2;
+        s.owned.wloc_pairs = 4000;
+        s.owned.wloc_bytes = 160_000;
         s.owned.h2h_keys = 50;
         s.owned.fence_runs = 10;
         s.bq_count = 4;
@@ -2484,7 +2492,7 @@ mod tests {
         assert!(line.contains("segs=3 sealed=2"), "{line}");
         assert!(line.contains("class_a=2000000"), "{line}");
         assert!(
-            line.contains("heap bq=32MiB iflight=3L/12000pin≈48MiB"),
+            line.contains("heap bq=32MiB iflight=3L/12000pin≈48MiB wloc=2L/4000pair≈0MiB"),
             "{line}"
         );
         assert!(!line.contains("union="), "{line}");

crates/rbitcoin-query/src/lib.rs

@@ -22,6 +22,7 @@ mod stamp;
 pub mod testutil;
 mod tx_precompute;
 mod wave_prevout;
+mod write_create_loc;
 
 #[cfg(debug_assertions)]
 pub use combined_stage::{body_ok_reads, reset_body_ok_reads};
@@ -95,17 +96,25 @@ pub struct ProcessOwnedSizes {
     pub fence_runs: usize,
     /// Body-queue heights whose raw payload was dropped after lookup decode.
     pub bq_promoted: usize,
+    /// Write-thread just-written `create.loc` packs (`wloc=`).
+    pub wloc_packs: usize,
+    pub wloc_pairs: usize,
+    pub wloc_bytes: u64,
 }
 
 /// Plan-thread published heap meters for structures not owned by [`Query`].
 ///
-/// Updated after each load note/prune ([`InFlight`]). Sampled by the ~5s IBD sizes line.
+/// Load publishes [`InFlight`] after note/prune; write publishes loc packs
+/// from TLS after note/prune. Sampled by the ~5s IBD sizes line.
 pub mod process_mem_stats {
     use std::sync::atomic::{AtomicU64, Ordering};
 
     static INFLIGHT_LAYERS: AtomicU64 = AtomicU64::new(0);
     static INFLIGHT_PINS: AtomicU64 = AtomicU64::new(0);
     static INFLIGHT_BYTES: AtomicU64 = AtomicU64::new(0);
+    static WLOC_PACKS: AtomicU64 = AtomicU64::new(0);
+    static WLOC_PAIRS: AtomicU64 = AtomicU64::new(0);
+    static WLOC_BYTES: AtomicU64 = AtomicU64::new(0);
 
     /// Publish latest prep-ahead occupancy (overwrite).
     pub fn note(inflight_layers: usize, inflight_pins: usize, inflight_bytes: u64) {
@@ -114,18 +123,31 @@ pub mod process_mem_stats {
         INFLIGHT_BYTES.store(inflight_bytes, Ordering::Relaxed);
     }
 
+    /// Write-thread loc window occupancy (TLS; no lock on Query).
+    pub fn note_wloc(packs: usize, pairs: usize, bytes: u64) {
+        WLOC_PACKS.store(packs as u64, Ordering::Relaxed);
+        WLOC_PAIRS.store(pairs as u64, Ordering::Relaxed);
+        WLOC_BYTES.store(bytes, Ordering::Relaxed);
+    }
+
     #[derive(Clone, Copy, Debug, Default)]
     pub struct Snap {
         pub inflight_layers: usize,
         pub inflight_pins: usize,
         pub inflight_bytes: u64,
+        pub wloc_packs: usize,
+        pub wloc_pairs: usize,
+        pub wloc_bytes: u64,
     }
 
     pub fn load() -> Snap {
         Snap {
             inflight_layers: INFLIGHT_LAYERS.load(Ordering::Relaxed) as usize,
             inflight_pins: INFLIGHT_PINS.load(Ordering::Relaxed) as usize,
             inflight_bytes: INFLIGHT_BYTES.load(Ordering::Relaxed),
+            wloc_packs: WLOC_PACKS.load(Ordering::Relaxed) as usize,
+            wloc_pairs: WLOC_PAIRS.load(Ordering::Relaxed) as usize,
+            wloc_bytes: WLOC_BYTES.load(Ordering::Relaxed),
         }
     }
 }
@@ -225,46 +247,6 @@ impl ShWriteBehind {
     }
 }
 
-/// Write-thread loc window (sequential fks). Cap drops the oldest pairs.
-#[derive(Default)]
-struct WriteCreateLocRam {
-    base: u64,
-    pairs: Vec<rbitcoin_store::CreateLocPair>,
-}
-
-impl WriteCreateLocRam {
-    const KEEP: usize = 1 << 20;
-
-    fn note(&mut self, fks: &[rbitcoin_primitives::Fk], loc: &[rbitcoin_store::CreateLocPair]) {
-        if fks.is_empty() || loc.len() != fks.len() {
-            return;
-        }
-        let Some(start) = fks[0].get() else {
-            return;
-        };
-        if self.pairs.is_empty() {
-            self.base = start;
-        }
-        let next = self.base.saturating_add(self.pairs.len() as u64);
-        if start != next {
-            self.base = start;
-            self.pairs.clear();
-        }
-        self.pairs.extend_from_slice(loc);
-        if self.pairs.len() > Self::KEEP {
-            let drop = self.pairs.len() - Self::KEEP;
-            self.pairs.drain(..drop);
-            self.base = self.base.saturating_add(drop as u64);
-        }
-    }
-
-    fn get(&self, fk: rbitcoin_primitives::Fk) -> Option<rbitcoin_store::CreateLocPair> {
-        let id = fk.get()?;
-        let off = id.checked_sub(self.base)?;
-        self.pairs.get(off as usize).copied()
-    }
-}
-
 /// Domain query facade used by higher layers (consensus, net, RPC).
 pub struct Query {
     store: Store,
@@ -292,9 +274,6 @@ pub struct Query {
     lookup_started_hi: AtomicU32,
     /// Max height whose Class A append committed (`u32::MAX` = none).
     class_a_hi: AtomicU32,
-    /// Write-thread loc pairs from Class A append. Later packs stamp abs from
-    /// this RAM (just-written parents). Write never preads `create.loc`.
-    write_create_loc: Mutex<WriteCreateLocRam>,
     /// Post-IBD SH SEAL + leftover-run discard (unsorted collect is tip finalize).
     sh_run: sh_builder::ShRunBuilder,
     /// Operator scripthash index intent (`--shindex`). When false, Class C skips
@@ -350,6 +329,7 @@ impl Query {
     }
 
     pub fn open_or_create_layout(layout: StoreLayout) -> Result<Self, QueryError> {
+        write_create_loc::clear();
         let store = Store::open_or_create_layout(layout)?;
         // Core checkblocks-style tip window first so repair sees the final fence.
         let reval = store.revalidate_tip_window()?;
@@ -400,7 +380,6 @@ impl Query {
             lookup_taken_hi: AtomicU32::new(u32::MAX),
             lookup_started_hi: AtomicU32::new(u32::MAX),
             class_a_hi: AtomicU32::new(u32::MAX),
-            write_create_loc: Mutex::new(WriteCreateLocRam::default()),
             sh_run: sh_builder::ShRunBuilder::new(&store_path),
             // Library default: SH on (tests / enter_direct). Node sets false for
             // `--shindex` off before entering Direct.
@@ -755,20 +734,41 @@ impl Query {
             .store(hi.unwrap_or(u32::MAX), AtomicOrdering::Release);
     }
 
-    /// Keep Class A append loc in RAM for later write packs (no loc pread).
+    /// Keep Class A append loc until write of the last pack whose TipOnly
+    /// may have missed it (`lookup_started_hi` at note), extended while the
+    /// pack is still at/above the load drain fence. Write-thread TLS. No loc pread.
     pub fn note_write_create_loc(
         &self,
         fks: &[rbitcoin_primitives::Fk],
         loc: &[rbitcoin_store::CreateLocPair],
+        pack_height: u32,
     ) {
-        self.write_create_loc.lock().unwrap().note(fks, loc);
+        let keep_until = self
+            .lookup_started_hi()
+            .unwrap_or(pack_height)
+            .max(pack_height);
+        write_create_loc::with_ram(self, |ram| {
+            ram.note(pack_height, keep_until, fks, loc);
+        });
+    }
+
+    /// Drop loc packs whose last overlapping lookup batch has finished write
+    /// and whose pack height is below the load drain fence.
+    pub fn prune_write_create_loc(&self, written_hi: u32) {
+        write_create_loc::with_ram(self, |ram| {
+            ram.prune_written_through(
+                written_hi,
+                self.lookup_started_hi(),
+                self.drain_and_fence_hi(),
+            );
+        });
     }
 
     pub fn write_create_loc(
         &self,
         fk: rbitcoin_primitives::Fk,
     ) -> Option<rbitcoin_store::CreateLocPair> {
-        self.write_create_loc.lock().unwrap().get(fk)
+        write_create_loc::with_ram(self, |ram| ram.get(fk))
     }
 
     /// Densify / offer: height is already in the confirm pipeline.
@@ -1056,6 +1056,9 @@ impl Query {
             h2h_keys,
             fence_runs: self.store.height_fence_run_count(),
             bq_promoted: self.block_queue_promoted_count(),
+            wloc_packs: mem.wloc_packs,
+            wloc_pairs: mem.wloc_pairs,
+            wloc_bytes: mem.wloc_bytes,
         }
     }