From aa641239e1715ea49a14373305c6a940fc404ea8 Mon Sep 17 00:00:00 2001 From: rbitcoin-grok Date: Sun, 13 Sep 2026 19:02:19 -0700 Subject: [PATCH 1/3] query: bound write loc packs to overlapping lookup write Just-written create.loc is invisible to TipOnly that already started. Keep those pairs until write of lookup_started_hi at note, with InFlight-style height packs, not a 1M FIFO. Disconnect drops the suffix. Co-authored-by: Cursor --- CHANGELOG.md | 10 +- SCHEMA.md | 5 +- .../src/confirm_run/write.rs | 12 +- .../src/confirm_run/write_idempotent_tests.rs | 47 +++- crates/rbitcoin-net/src/ibd/perf_log.rs | 12 +- crates/rbitcoin-query/src/lib.rs | 85 +++--- crates/rbitcoin-query/src/query_tests.rs | 16 ++ crates/rbitcoin-query/src/write_create_loc.rs | 242 ++++++++++++++++++ docs/ibd-memory.md | 3 +- docs/invariants.md | 4 +- 10 files changed, 377 insertions(+), 59 deletions(-) create mode 100644 crates/rbitcoin-query/src/write_create_loc.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 90a473955..48067cc61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -108,10 +108,12 @@ 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 packs until write of `lookup_started_hi` at note (last + TipOnly that may have missed disk loc). Same-batch / just-written abs + stamp from that RAM (packed pin outs). Disconnect drops packs at/above + that height. 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 diff --git a/SCHEMA.md b/SCHEMA.md index fb1ef0cae..05e8c8cd8 100644 --- a/SCHEMA.md +++ b/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. diff --git a/crates/rbitcoin-consensus/src/confirm_run/write.rs b/crates/rbitcoin-consensus/src/confirm_run/write.rs index 9a1d3fc24..1fd05125b 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/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, diff --git a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs index 455a47f5e..d7af761dd 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs +++ b/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" + ); + 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); } diff --git a/crates/rbitcoin-net/src/ibd/perf_log.rs b/crates/rbitcoin-net/src/ibd/perf_log.rs index d3eefe8ba..7a0a17e28 100644 --- a/crates/rbitcoin-net/src/ibd/perf_log.rs +++ b/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}"); diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 06a0663d6..698b5dbff 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/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}; @@ -47,7 +48,7 @@ use rbitcoin_store::{ script_hash, HeaderRecord, InputRecord, OutputRecord, PointRecord, ScriptHashRecord, SpTweaksTable, Store, StoreError, StoreLayout, TxRecord, }; -use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering as AtomicOrdering}; use std::sync::{Arc, Condvar, Mutex}; @@ -95,6 +96,10 @@ 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`]. @@ -225,46 +230,6 @@ impl ShWriteBehind { } } -/// Write-thread loc window (sequential fks). Cap drops the oldest pairs. -#[derive(Default)] -struct WriteCreateLocRam { - base: u64, - pairs: Vec, -} - -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 { - 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, @@ -294,7 +259,8 @@ pub struct Query { 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, + /// Packs live until write of `lookup_started_hi` at note time. + write_create_loc: Mutex, /// 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 @@ -400,7 +366,7 @@ 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()), + write_create_loc: Mutex::new(write_create_loc::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. @@ -473,6 +439,10 @@ impl Query { self.set_lookup_taken_hi(rewind); self.set_lookup_started_hi(rewind); self.set_class_a_hi(rewind); + self.write_create_loc + .lock() + .unwrap() + .drop_from_height(height); self.block_queue_drop_resolved_from(height); } @@ -755,13 +725,30 @@ 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). 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); + self.write_create_loc + .lock() + .unwrap() + .note(pack_height, keep_until, fks, loc); + } + + /// Drop loc packs whose last overlapping lookup batch has finished write. + pub fn prune_write_create_loc(&self, written_hi: u32) { + self.write_create_loc + .lock() + .unwrap() + .prune_written_through(written_hi); } pub fn write_create_loc( @@ -1034,6 +1021,11 @@ impl Query { // Wire path always put_header_plan; conf_plans=0 was a metering bug. let conf_plans = self.confirm_parents.header_plan_count(); let mem = process_mem_stats::load(); + let (wloc_packs, wloc_pairs, wloc_bytes) = self + .write_create_loc + .lock() + .unwrap_or_else(|e| e.into_inner()) + .size_snapshot(); let h2h_keys = self .height_by_hash .lock() @@ -1056,6 +1048,9 @@ impl Query { h2h_keys, fence_runs: self.store.height_fence_run_count(), bq_promoted: self.block_queue_promoted_count(), + wloc_packs, + wloc_pairs, + wloc_bytes, } } diff --git a/crates/rbitcoin-query/src/query_tests.rs b/crates/rbitcoin-query/src/query_tests.rs index ad7db8284..c1d5367ad 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -178,10 +178,26 @@ fn note_disconnect_rewinds_started_and_class_a_with_taken() { q.set_lookup_taken_hi(Some(12)); q.set_lookup_started_hi(Some(12)); q.set_class_a_hi(Some(10)); + let fk = Fk(10); + let pair = rbitcoin_store::CreateLocPair { + txout: (10, 8), + spent: (20, 8), + n_out: 1, + }; + q.note_write_create_loc(&[fk], &[pair], 10); + assert!(q.write_create_loc(fk).is_some()); + let snap = q.process_owned_size_snapshot(); + assert_eq!(snap.wloc_packs, 1); + assert_eq!(snap.wloc_pairs, 1); + assert!(snap.wloc_bytes > 0); q.note_disconnect_height(8); assert_eq!(q.lookup_taken_hi(), Some(7)); assert_eq!(q.lookup_started_hi(), Some(7)); assert_eq!(q.class_a_hi(), Some(7)); + assert!( + q.write_create_loc(fk).is_none(), + "disconnect drops write loc packs at/above that height" + ); let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-query/src/write_create_loc.rs b/crates/rbitcoin-query/src/write_create_loc.rs new file mode 100644 index 000000000..e18bb9187 --- /dev/null +++ b/crates/rbitcoin-query/src/write_create_loc.rs @@ -0,0 +1,242 @@ +//! Write-thread loc pairs from Class A append (just-written parents). +//! +//! Lookup TipOnly reads `create.loc` after [`crate::Query::note_lookup_tiponly_start`]. +//! A later pack whose TipOnly already started cannot have stamped those pairs. +//! Write keeps them until that last pack has finished write: `keep_until` is +//! `lookup_started_hi` at note (at least the noting pack height). Prune when +//! written height ≥ `keep_until`. Disconnect [`Self::drop_from_height`] drops +//! packs at/above that height and clamps remaining `keep_until`. Height index +//! matches [`crate::InFlight`]; the drop gate is written height vs keep-until, +//! not drain+fence vs pack height. + +use rbitcoin_primitives::Fk; +use rbitcoin_store::CreateLocPair; +use std::collections::BTreeMap; + +const PAIR_BYTES: u64 = std::mem::size_of::() as u64; + +#[derive(Debug)] +struct LocPack { + keep_until: u32, + base: u64, + pairs: Vec, + approx_bytes: u64, +} + +impl LocPack { + fn get(&self, id: u64) -> Option { + let off = id.checked_sub(self.base)?; + self.pairs.get(off as usize).copied() + } +} + +/// Write-owned loc window. Note after append; prune after that write batch +/// (and later overlapping batches) finish. +#[derive(Debug, Default)] +pub(crate) struct WriteCreateLocRam { + by_height: BTreeMap, + approx_bytes: u64, +} + +impl WriteCreateLocRam { + pub(crate) fn note( + &mut self, + pack_height: u32, + keep_until: u32, + fks: &[Fk], + loc: &[CreateLocPair], + ) { + if fks.is_empty() || loc.len() != fks.len() { + return; + } + let Some(start) = fks[0].get() else { + return; + }; + let bytes = (loc.len() as u64).saturating_mul(PAIR_BYTES); + if let Some(old) = self.by_height.insert( + pack_height, + LocPack { + keep_until, + base: start, + pairs: loc.to_vec(), + approx_bytes: bytes, + }, + ) { + self.approx_bytes = self.approx_bytes.saturating_sub(old.approx_bytes); + } + self.approx_bytes = self.approx_bytes.saturating_add(bytes); + } + + pub(crate) fn get(&self, fk: Fk) -> Option { + let id = fk.get()?; + for pack in self.by_height.values() { + if let Some(p) = pack.get(id) { + return Some(p); + } + } + None + } + + /// Drop packs whose last overlapping lookup batch has finished write. + /// + /// Equality drops (`keep_until == written_hi`). + pub(crate) fn prune_written_through(&mut self, written_hi: u32) { + let drop: Vec = self + .by_height + .iter() + .filter(|(_, p)| p.keep_until <= written_hi) + .map(|(h, _)| *h) + .collect(); + for h in drop { + if let Some(p) = self.by_height.remove(&h) { + self.approx_bytes = self.approx_bytes.saturating_sub(p.approx_bytes); + } + } + } + + /// Disconnect: drop packs at/above `height`. Remaining keep-until cannot + /// wait for disconnected lookup batches. + pub(crate) fn drop_from_height(&mut self, height: u32) { + let drop = self.by_height.split_off(&height); + for p in drop.into_values() { + self.approx_bytes = self.approx_bytes.saturating_sub(p.approx_bytes); + } + let cap = height.saturating_sub(1); + for p in self.by_height.values_mut() { + if p.keep_until > cap { + p.keep_until = cap; + } + } + } + + /// Packs, pair count, approx bytes (IBD `wloc=`). + pub(crate) fn size_snapshot(&self) -> (usize, usize, u64) { + let pairs: usize = self.by_height.values().map(|p| p.pairs.len()).sum(); + (self.by_height.len(), pairs, self.approx_bytes) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rbitcoin_primitives::Fk; + + fn pair(n: u64) -> CreateLocPair { + CreateLocPair { + txout: (n * 10, 8), + spent: (n * 9, 8), + n_out: 1, + } + } + + fn fks(ids: &[u64]) -> Vec { + ids.iter().copied().map(Fk).collect() + } + + fn loc(ids: &[u64]) -> Vec { + ids.iter().copied().map(pair).collect() + } + + #[test] + fn note_get_by_fk() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &fks(&[10, 11]), &loc(&[10, 11])); + assert_eq!(m.get(Fk(10)).unwrap().txout, (100, 8)); + assert_eq!(m.get(Fk(11)).unwrap().spent, (99, 8)); + assert!(m.get(Fk(12)).is_none()); + assert!(m.get(Fk(9)).is_none()); + } + + #[test] + fn prune_written_through_drops_at_keep_until() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &fks(&[10]), &loc(&[10])); + m.prune_written_through(3); + assert!(m.get(Fk(10)).is_some(), "written_hi < keep_until keeps"); + m.prune_written_through(4); + assert!( + m.get(Fk(10)).is_none(), + "written_hi == keep_until drops (last overlapping batch finished write)" + ); + assert_eq!(m.size_snapshot(), (0, 0, 0)); + } + + #[test] + fn prune_written_through_is_per_pack_keep_until() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &fks(&[10]), &loc(&[10])); + m.note(2, 8, &fks(&[11]), &loc(&[11])); + m.prune_written_through(4); + assert!(m.get(Fk(10)).is_none()); + assert!(m.get(Fk(11)).is_some(), "later pack waits for keep_until 8"); + m.prune_written_through(8); + assert!(m.get(Fk(11)).is_none()); + } + + #[test] + fn no_count_cap_while_keep_until_open() { + let mut m = WriteCreateLocRam::default(); + for i in 0..8u64 { + m.note((i + 1) as u32, 100, &[Fk(1000 + i)], &[pair(1000 + i)]); + } + let (packs, pairs, bytes) = m.size_snapshot(); + assert_eq!(packs, 8); + assert_eq!(pairs, 8); + assert_eq!(bytes, 8 * PAIR_BYTES); + assert!(m.get(Fk(1000)).is_some()); + assert!(m.get(Fk(1007)).is_some()); + m.prune_written_through(99); + assert_eq!(m.size_snapshot().0, 8, "no silent FIFO drop"); + m.prune_written_through(100); + assert_eq!(m.size_snapshot(), (0, 0, 0)); + } + + #[test] + fn drop_from_height_drops_suffix_and_clamps() { + let mut m = WriteCreateLocRam::default(); + m.note(5, 20, &fks(&[10]), &loc(&[10])); + m.note(8, 20, &fks(&[11]), &loc(&[11])); + m.drop_from_height(8); + assert!(m.get(Fk(10)).is_some()); + assert!(m.get(Fk(11)).is_none()); + m.prune_written_through(7); + assert!( + m.get(Fk(10)).is_none(), + "keep_until clamped to disconnect-1 so remaining pipeline can retire the pack" + ); + } + + #[test] + fn drop_from_height_zero_clears() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &fks(&[10]), &loc(&[10])); + m.drop_from_height(0); + assert!(m.get(Fk(10)).is_none()); + assert_eq!(m.size_snapshot(), (0, 0, 0)); + } + + #[test] + fn empty_or_len_mismatch_is_noop() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &[], &[]); + m.note(1, 4, &fks(&[10]), &loc(&[10, 11])); + assert_eq!(m.size_snapshot(), (0, 0, 0)); + m.note(1, 4, &fks(&[0]), &loc(&[0])); + assert!(m.get(Fk(0)).is_none(), "null fk does not note"); + } + + #[test] + fn same_pack_height_replace_does_not_double_count_bytes() { + let mut m = WriteCreateLocRam::default(); + m.note(1, 4, &fks(&[10, 11]), &loc(&[10, 11])); + let (_, _, once) = m.size_snapshot(); + m.note(1, 8, &fks(&[10]), &loc(&[10])); + let (packs, pairs, twice) = m.size_snapshot(); + assert_eq!(packs, 1); + assert_eq!(pairs, 1); + assert_eq!(twice, PAIR_BYTES); + assert!(twice < once); + assert!(m.get(Fk(11)).is_none()); + assert!(m.get(Fk(10)).is_some()); + } +} diff --git a/docs/ibd-memory.md b/docs/ibd-memory.md index c3ad1aef7..a09508514 100644 --- a/docs/ibd-memory.md +++ b/docs/ibd-memory.md @@ -35,6 +35,7 @@ pres and **not** the raw bytes. Reorg gather that wants wire re-encodes. | **Load-batch parent skeleton** | Wave `txid → (fk, body_range)` + spent ranges on the `LoadBatch`; per-chunk need-vouts at split | Drop with the batch. Lookup TipOnly only. Not a process FIFO. | | **Pipeline pins (no process FIFO)** | Plan `batch_pin` / `BatchParents` only | Drop with batch. Cold **outs** for ancient parents use `txout.body` into `BatchParents` (stamped range). Recent-window first spend uses stamp-carried in-flight CreatePin until load drops the pack below the wave's pre-TipOnly drain+fence snapshot | | **In-flight CreatePin map** | identity + full create outs; one load-thread HashMap; lookup snapshots `drain_and_fence_hi` before TipOnly and passes it on the last load batch; load drops pack height below that after the in-flight read | Load notes after stamp; disconnect `drop_from` on pack height. Sizes: `iflight=`. Not a coins cache / spend FIFO | +| **Write loc packs** | Class A append `CreateLocPair`s on the write thread; height-tagged like InFlight. `keep_until` = `lookup_started_hi` at note (last TipOnly that may have missed disk loc). No count cap | Write notes after append; prune when that write batch (and later overlapping batches) finish (`written_hi ≥ keep_until`). Disconnect `drop_from` on pack height and clamps remaining keep-until. Sizes: `wloc=` | | **ConfirmParentCache header plans** | tip-GC window | Always on — required for multi-block wire MTP | | **Confirm plans / headers** | offer-ahead window | `ConfirmParentCache::advance_tip` from write `post_commit` | | **SH catalog runs** | leftover `scripthash.runs` discarded at tip (unsorted collect does not write them) | write-behind / discard; not during Direct confirm | @@ -126,7 +127,7 @@ grep 'tip: perf' mainnet.log | `conf loadq=` / `scriptq` / `writeq` | Real queue contents (loadq cap **14**) + pipeline-wide `parents=` + feed ready/inflight | | `txhead` | Segmented `tx.head.*` (open head + sealed heads/fuses; logical sizes) | | `sh` | SH catalog runs / tip heads | -| `heap … iflight= h2h= fence= fuse8= mphf_g= open_keys= class_c_l2= accounted= residual=` | Approx process heap: BQ + load-ahead CreatePins (`iflight=`) + `height_by_hash` + height fence (`Arc` snapshot for leftover TipOnly — not a 15 MiB memcpy/wave) + confirm wire + **sealed `tx.head` fuse8 fingerprints** + FdOnly BDZ `g` heap (`mphf_g=`, 0 after open) + open-segment fuse-key Vec + Class C L2 images; residual = anon − accounted | +| `heap … iflight= wloc= h2h= fence= fuse8= mphf_g= open_keys= class_c_l2= accounted= residual=` | Approx process heap: BQ + load-ahead CreatePins (`iflight=`) + write loc packs (`wloc=`) + `height_by_hash` + height fence (`Arc` snapshot for leftover TipOnly — not a 15 MiB memcpy/wave) + confirm wire + **sealed `tx.head` fuse8 fingerprints** + FdOnly BDZ `g` heap (`mphf_g=`, 0 after open) + open-segment fuse-key Vec + Class C L2 images; residual = anon − accounted | ## Residual heap audit (872k / ~1.42 B creates) diff --git a/docs/invariants.md b/docs/invariants.md index 335353ca3..c135b8b56 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -82,7 +82,7 @@ not head/idx. | Parent create_fk | **same-batch** planned fks (offline at pin) → **in-flight** (lookup snapshots `drain_and_fence_hi` **before** the wave's TipOnly read and passes it on the last load batch; load drops tagged map rows with pack height **below** that snapshot after that batch's in-flight read; equality keeps; not Class C tip, not `class_a_hi`, not write freeze; one load-thread HashMap, insert after stamp) → **skeleton** (`BatchParentIds` on the `LoadBatch`: lookup TipOnly fk + body_range + spent_range + per-chunk need-vouts) → **Corrupt** on IBD miss. plan=None / S0 (`skeleton = None`) is in-flight → leftover TipOnly. One helper: [`stamp_external_parents`](../crates/rbitcoin-query/src/stamp.rs). No leftover pending map, no process pin FIFO, no BQ-side hits map, no parent-store create_fk on stamp, no published live_union. Same-wave creates are omitted from TipOnly need. Header-cache GC polls store tip each load pack. One fk per txid — [`errata.md`](./errata.md). | Miss of in-flight and skeleton → `Corrupt("parent create_fk unresolved")` (**engine fault**: requeue once, then halt IBD; never blacklist). Identity without loc range → `Corrupt("invariant: loc range missing after identity")`, not a miss | | io_uring harvest | TLS session fail-closed ([`io-modality.md`](./io-modality.md)) | **No** silent success. `Corrupt("invariant: io_uring …")` (not `bdz g page bad slot`). Ring-unavailable still pread-fallback | | Load body outs | By `txout` range only from lookup stamp; incomplete outs → hard Err. Pin **copies** lookup `spent_range` (no loc IO) | **No** loc cold outs on load; **no** `create.loc` on load; **no** `inwit` on pin | -| Ensure (write) | Every non-null spend edge has `spent_range` abs after ensure returns. Lookup stamped archived parents; same-batch and just-written packs from Class A append RAM (packed pin outs / write loc window). Missing abs → `invariant:` | **No** write `create.loc` read | +| Ensure (write) | Every non-null spend edge has `spent_range` abs after ensure returns. Lookup stamped archived parents; same-batch from append RAM loc; just-written from write loc packs kept until write of `lookup_started_hi` at note (last TipOnly that may have missed disk loc). Missing abs → `invariant:` | **No** write `create.loc` read | | Structural spentness | Abs required for every non-null spend create_fk after load; multi-list → confirmed-strong walk (reorg protocol) | **No** unpinned “wire-corrected create_fk” soft spentness. Multi flag alone is **not** hard `Err` | | Pin create identity | Pin must carry non-zero create txid from **lookup stamp** (plan reverse map / wire prev_txid / `txid.body`) | Soft zero-identity pin → assemble mismatch → cold recovery is **forbidden** | | Tip already-archived | `plan=None`: lookup still stamps parent pin material; load `txout` by range | Soft spentness recovery for zero pin identity is **not** OK | @@ -162,7 +162,7 @@ packs at/above the leaving **pack** height **before** the next bind. | `pin_for_wire_incomplete_outs_is_invariant_error` | `pin_for_wire_batch` incomplete outs → cold miss | | `post_commit_missing_denserels_is_invariant_error` | `post_commit` abs-only annotate | | `ensure_spend_abs_incomplete_is_invariant_error` | `ensure_spend_abs_layouts` post-condition | -| `write_ensure_stamps_spent_range_after_load_pin` / `pin_and_ensure_journey` / `fill_same_batch_abs_from_append_loc_ram` | load pin copies lookup spent range; missing stamp is Corrupt; same-batch abs from append RAM (no write loc pread) | +| `write_ensure_stamps_spent_range_after_load_pin` / `pin_and_ensure_journey` / `fill_same_batch_abs_from_append_loc_ram` | load pin copies lookup spent range; missing stamp is Corrupt; same-batch abs from append RAM; just-written loc until write of `lookup_started_hi` (no write loc pread) | | `fill_missing_parent_ranges_stamps_spent_idx_for_archived` | lookup stamp carries spent range for TipOnly leftover | | `spend_abs_jobs_unique_and_missing_is_corrupt` | pin arithmetic abs list; missing → Corrupt | | `structural_pinned_without_abs_is_invariant_error` | `structural_validate_spends` pin without denserels | From 0a69d565f78bc5661f5a75b4dd0b3864a7193e45 Mon Sep 17 00:00:00 2001 From: rbitcoin-grok Date: Sun, 13 Sep 2026 19:05:28 -0700 Subject: [PATCH 2/3] query: restore BTreeMap import for super glob scripthash.rs uses BTreeMap via use super::*. Dropping the crate-root use broke clippy/test compile. Co-authored-by: Cursor --- crates/rbitcoin-query/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 698b5dbff..f68bbd980 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -48,7 +48,7 @@ use rbitcoin_store::{ script_hash, HeaderRecord, InputRecord, OutputRecord, PointRecord, ScriptHashRecord, SpTweaksTable, Store, StoreError, StoreLayout, TxRecord, }; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering as AtomicOrdering}; use std::sync::{Arc, Condvar, Mutex}; From 5090d0fae1a79b6fe7c367fe432ed8fb14fc6603 Mon Sep 17 00:00:00 2001 From: rbitcoin-grok Date: Sun, 13 Sep 2026 19:28:17 -0700 Subject: [PATCH 3/3] query: write loc is TLS; keep noting pack past same-write prune Query mutex was leftover parking for a write-only map. Move packs to write-thread TLS (atomics for wloc=) and poll disconnect there. Do not drop the noting pack until a later write below the load drain fence so a child whose TipOnly skipped loc via InFlight can still fill abs. Co-authored-by: Cursor --- CHANGELOG.md | 11 +- .../src/confirm_run/write_idempotent_tests.rs | 2 +- crates/rbitcoin-query/src/lib.rs | 66 ++++++---- crates/rbitcoin-query/src/query_tests.rs | 4 - crates/rbitcoin-query/src/write_create_loc.rs | 124 +++++++++++++++--- docs/ibd-memory.md | 2 +- docs/invariants.md | 4 +- 7 files changed, 154 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 48067cc61..cbf1499ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -109,11 +109,12 @@ before 1.0). snapshot. Live table drops before cleanup. - **Write loc RAM:** Class A append returns loc pairs; write keeps them in - height-tagged packs until write of `lookup_started_hi` at note (last - TipOnly that may have missed disk loc). Same-batch / just-written abs - stamp from that RAM (packed pin outs). Disconnect drops packs at/above - that height. Write does not pread `create.loc`. Missing stamp is - `Corrupt`. `ibd: sizes` `wloc=`. + 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 diff --git a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs index d7af761dd..30d5facc1 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs @@ -1312,7 +1312,7 @@ fn fill_same_batch_abs_from_append_loc_ram() { q.prune_write_create_loc(3); assert!( q.write_create_loc(fks[0]).is_some(), - "keep until write of lookup_started_hi" + "keep until write of lookup_started_hi (and pack below drain fence)" ); q.prune_write_create_loc(4); assert!( diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index f68bbd980..accde7352 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -104,13 +104,17 @@ pub struct ProcessOwnedSizes { /// 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) { @@ -119,11 +123,21 @@ 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 { @@ -131,6 +145,9 @@ pub mod process_mem_stats { 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), } } } @@ -257,10 +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`. - /// Packs live until write of `lookup_started_hi` at note time. - write_create_loc: Mutex, /// 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 @@ -316,6 +329,7 @@ impl Query { } pub fn open_or_create_layout(layout: StoreLayout) -> Result { + 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()?; @@ -366,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(write_create_loc::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. @@ -439,10 +452,6 @@ impl Query { self.set_lookup_taken_hi(rewind); self.set_lookup_started_hi(rewind); self.set_class_a_hi(rewind); - self.write_create_loc - .lock() - .unwrap() - .drop_from_height(height); self.block_queue_drop_resolved_from(height); } @@ -726,7 +735,8 @@ impl Query { } /// Keep Class A append loc until write of the last pack whose TipOnly - /// may have missed it (`lookup_started_hi` at note). No loc pread. + /// 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], @@ -737,25 +747,28 @@ impl Query { .lookup_started_hi() .unwrap_or(pack_height) .max(pack_height); - self.write_create_loc - .lock() - .unwrap() - .note(pack_height, keep_until, fks, loc); + 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. + /// 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) { - self.write_create_loc - .lock() - .unwrap() - .prune_written_through(written_hi); + 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 { - 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. @@ -1021,11 +1034,6 @@ impl Query { // Wire path always put_header_plan; conf_plans=0 was a metering bug. let conf_plans = self.confirm_parents.header_plan_count(); let mem = process_mem_stats::load(); - let (wloc_packs, wloc_pairs, wloc_bytes) = self - .write_create_loc - .lock() - .unwrap_or_else(|e| e.into_inner()) - .size_snapshot(); let h2h_keys = self .height_by_hash .lock() @@ -1048,9 +1056,9 @@ impl Query { h2h_keys, fence_runs: self.store.height_fence_run_count(), bq_promoted: self.block_queue_promoted_count(), - wloc_packs, - wloc_pairs, - wloc_bytes, + wloc_packs: mem.wloc_packs, + wloc_pairs: mem.wloc_pairs, + wloc_bytes: mem.wloc_bytes, } } diff --git a/crates/rbitcoin-query/src/query_tests.rs b/crates/rbitcoin-query/src/query_tests.rs index c1d5367ad..0e4db7de3 100644 --- a/crates/rbitcoin-query/src/query_tests.rs +++ b/crates/rbitcoin-query/src/query_tests.rs @@ -186,10 +186,6 @@ fn note_disconnect_rewinds_started_and_class_a_with_taken() { }; q.note_write_create_loc(&[fk], &[pair], 10); assert!(q.write_create_loc(fk).is_some()); - let snap = q.process_owned_size_snapshot(); - assert_eq!(snap.wloc_packs, 1); - assert_eq!(snap.wloc_pairs, 1); - assert!(snap.wloc_bytes > 0); q.note_disconnect_height(8); assert_eq!(q.lookup_taken_hi(), Some(7)); assert_eq!(q.lookup_started_hi(), Some(7)); diff --git a/crates/rbitcoin-query/src/write_create_loc.rs b/crates/rbitcoin-query/src/write_create_loc.rs index e18bb9187..052ebfdb5 100644 --- a/crates/rbitcoin-query/src/write_create_loc.rs +++ b/crates/rbitcoin-query/src/write_create_loc.rs @@ -3,16 +3,23 @@ //! Lookup TipOnly reads `create.loc` after [`crate::Query::note_lookup_tiponly_start`]. //! A later pack whose TipOnly already started cannot have stamped those pairs. //! Write keeps them until that last pack has finished write: `keep_until` is -//! `lookup_started_hi` at note (at least the noting pack height). Prune when -//! written height ≥ `keep_until`. Disconnect [`Self::drop_from_height`] drops -//! packs at/above that height and clamps remaining `keep_until`. Height index -//! matches [`crate::InFlight`]; the drop gate is written height vs keep-until, -//! not drain+fence vs pack height. +//! `lookup_started_hi` at note (at least the noting pack height), then extended +//! while the pack is still at/above the load drain fence (TipOnly still skips +//! disk loc via InFlight). Prune when pack height is below that fence **and** +//! written height ≥ `keep_until`. The noting pack is not dropped on the same +//! write (`pack_height < written_hi`). +//! +//! This window is **write-thread TLS** (one writer per thread — same ownership +//! as load's [`crate::InFlight`], not a `Query` mutex). Disconnect is polled +//! via [`crate::Query::take_disconnect`] on the write thread. use rbitcoin_primitives::Fk; use rbitcoin_store::CreateLocPair; +use std::cell::{Cell, RefCell}; use std::collections::BTreeMap; +use crate::Query; + const PAIR_BYTES: u64 = std::mem::size_of::() as u64; #[derive(Debug)] @@ -77,14 +84,22 @@ impl WriteCreateLocRam { None } - /// Drop packs whose last overlapping lookup batch has finished write. - /// - /// Equality drops (`keep_until == written_hi`). - pub(crate) fn prune_written_through(&mut self, written_hi: u32) { + /// Drop packs whose loc is on disk for later TipOnly (`pack_height < fence`) + /// and whose overlapping lookup wave has finished write (`keep_until <= + /// written_hi`). Keep the noting pack so a child that looked up while the + /// parent was still in-flight can still fill abs. + pub(crate) fn prune_written_through( + &mut self, + written_hi: u32, + started_hi: Option, + fence_hi: Option, + ) { + let started = started_hi.unwrap_or(written_hi); + let fence = fence_hi.unwrap_or(written_hi); let drop: Vec = self .by_height .iter() - .filter(|(_, p)| p.keep_until <= written_hi) + .filter(|(h, p)| **h < fence && **h < written_hi && p.keep_until <= written_hi) .map(|(h, _)| *h) .collect(); for h in drop { @@ -92,6 +107,11 @@ impl WriteCreateLocRam { self.approx_bytes = self.approx_bytes.saturating_sub(p.approx_bytes); } } + for (h, pack) in self.by_height.iter_mut() { + if *h >= fence { + pack.keep_until = pack.keep_until.max(started); + } + } } /// Disconnect: drop packs at/above `height`. Remaining keep-until cannot @@ -116,6 +136,43 @@ impl WriteCreateLocRam { } } +thread_local! { + static RAM: RefCell = RefCell::new(WriteCreateLocRam::default()); + static DISCONNECT_SEEN: Cell = const { Cell::new(0) }; +} + +fn publish(ram: &WriteCreateLocRam) { + let (packs, pairs, bytes) = ram.size_snapshot(); + crate::process_mem_stats::note_wloc(packs, pairs, bytes); +} + +/// Drop this thread's loc window (Query open / test isolation). Does not publish +/// zeros — `wloc=` atomics are process-wide like `iflight=`. +pub(crate) fn clear() { + RAM.with(|c| { + *c.borrow_mut() = WriteCreateLocRam::default(); + }); + DISCONNECT_SEEN.with(|c| c.set(0)); +} + +pub(crate) fn with_ram(query: &Query, f: impl FnOnce(&mut WriteCreateLocRam) -> R) -> R { + RAM.with(|c| { + let mut ram = c.borrow_mut(); + poll_disconnect(query, &mut ram); + let r = f(&mut ram); + publish(&ram); + r + }) +} + +fn poll_disconnect(query: &Query, ram: &mut WriteCreateLocRam) { + let mut seen = DISCONNECT_SEEN.with(|c| c.get()); + if let Some(h) = query.take_disconnect(&mut seen) { + ram.drop_from_height(h); + DISCONNECT_SEEN.with(|c| c.set(seen)); + } +} + #[cfg(test)] mod tests { use super::*; @@ -137,6 +194,10 @@ mod tests { ids.iter().copied().map(pair).collect() } + fn prune(m: &mut WriteCreateLocRam, written_hi: u32) { + m.prune_written_through(written_hi, Some(written_hi), Some(written_hi)); + } + #[test] fn note_get_by_fk() { let mut m = WriteCreateLocRam::default(); @@ -151,9 +212,9 @@ mod tests { fn prune_written_through_drops_at_keep_until() { let mut m = WriteCreateLocRam::default(); m.note(1, 4, &fks(&[10]), &loc(&[10])); - m.prune_written_through(3); + prune(&mut m, 3); assert!(m.get(Fk(10)).is_some(), "written_hi < keep_until keeps"); - m.prune_written_through(4); + prune(&mut m, 4); assert!( m.get(Fk(10)).is_none(), "written_hi == keep_until drops (last overlapping batch finished write)" @@ -166,13 +227,42 @@ mod tests { let mut m = WriteCreateLocRam::default(); m.note(1, 4, &fks(&[10]), &loc(&[10])); m.note(2, 8, &fks(&[11]), &loc(&[11])); - m.prune_written_through(4); + prune(&mut m, 4); assert!(m.get(Fk(10)).is_none()); assert!(m.get(Fk(11)).is_some(), "later pack waits for keep_until 8"); - m.prune_written_through(8); + prune(&mut m, 8); assert!(m.get(Fk(11)).is_none()); } + #[test] + fn prune_keeps_noting_pack_until_later_write() { + let mut m = WriteCreateLocRam::default(); + m.note(3, 3, &fks(&[10]), &loc(&[10])); + prune(&mut m, 3); + assert!( + m.get(Fk(10)).is_some(), + "same-write prune must not drop the noting pack (child TipOnly may still skip loc via InFlight)" + ); + prune(&mut m, 4); + assert!(m.get(Fk(10)).is_none()); + assert_eq!(m.size_snapshot(), (0, 0, 0)); + } + + #[test] + fn prune_bumps_keep_until_from_started_hi_while_at_fence() { + let mut m = WriteCreateLocRam::default(); + m.note(2, 1, &fks(&[10]), &loc(&[10])); + m.prune_written_through(2, Some(8), Some(2)); + assert!(m.get(Fk(10)).is_some()); + m.prune_written_through(7, Some(8), Some(9)); + assert!( + m.get(Fk(10)).is_some(), + "keep_until bumped to 8 while pack was still at fence" + ); + m.prune_written_through(8, Some(8), Some(9)); + assert!(m.get(Fk(10)).is_none()); + } + #[test] fn no_count_cap_while_keep_until_open() { let mut m = WriteCreateLocRam::default(); @@ -185,9 +275,9 @@ mod tests { assert_eq!(bytes, 8 * PAIR_BYTES); assert!(m.get(Fk(1000)).is_some()); assert!(m.get(Fk(1007)).is_some()); - m.prune_written_through(99); + prune(&mut m, 99); assert_eq!(m.size_snapshot().0, 8, "no silent FIFO drop"); - m.prune_written_through(100); + prune(&mut m, 100); assert_eq!(m.size_snapshot(), (0, 0, 0)); } @@ -199,7 +289,7 @@ mod tests { m.drop_from_height(8); assert!(m.get(Fk(10)).is_some()); assert!(m.get(Fk(11)).is_none()); - m.prune_written_through(7); + prune(&mut m, 7); assert!( m.get(Fk(10)).is_none(), "keep_until clamped to disconnect-1 so remaining pipeline can retire the pack" diff --git a/docs/ibd-memory.md b/docs/ibd-memory.md index a09508514..ebcb9c7b5 100644 --- a/docs/ibd-memory.md +++ b/docs/ibd-memory.md @@ -35,7 +35,7 @@ pres and **not** the raw bytes. Reorg gather that wants wire re-encodes. | **Load-batch parent skeleton** | Wave `txid → (fk, body_range)` + spent ranges on the `LoadBatch`; per-chunk need-vouts at split | Drop with the batch. Lookup TipOnly only. Not a process FIFO. | | **Pipeline pins (no process FIFO)** | Plan `batch_pin` / `BatchParents` only | Drop with batch. Cold **outs** for ancient parents use `txout.body` into `BatchParents` (stamped range). Recent-window first spend uses stamp-carried in-flight CreatePin until load drops the pack below the wave's pre-TipOnly drain+fence snapshot | | **In-flight CreatePin map** | identity + full create outs; one load-thread HashMap; lookup snapshots `drain_and_fence_hi` before TipOnly and passes it on the last load batch; load drops pack height below that after the in-flight read | Load notes after stamp; disconnect `drop_from` on pack height. Sizes: `iflight=`. Not a coins cache / spend FIFO | -| **Write loc packs** | Class A append `CreateLocPair`s on the write thread; height-tagged like InFlight. `keep_until` = `lookup_started_hi` at note (last TipOnly that may have missed disk loc). No count cap | Write notes after append; prune when that write batch (and later overlapping batches) finish (`written_hi ≥ keep_until`). Disconnect `drop_from` on pack height and clamps remaining keep-until. Sizes: `wloc=` | +| **Write loc packs** | Class A append `CreateLocPair`s on the **write thread** (TLS, no `Query` mutex); height-tagged like InFlight. `keep_until` = `lookup_started_hi` at note, then extended while pack height ≥ load drain fence (TipOnly still skips disk loc via InFlight). No count cap | Write notes after append; prune when pack height is below that fence **and** `written_hi ≥ keep_until`. Same-write prune keeps the noting pack. Disconnect is polled on the write thread (`take_disconnect`) and `drop_from` on pack height (clamps remaining keep-until). Sizes: `wloc=` (atomics, like `iflight=`) | | **ConfirmParentCache header plans** | tip-GC window | Always on — required for multi-block wire MTP | | **Confirm plans / headers** | offer-ahead window | `ConfirmParentCache::advance_tip` from write `post_commit` | | **SH catalog runs** | leftover `scripthash.runs` discarded at tip (unsorted collect does not write them) | write-behind / discard; not during Direct confirm | diff --git a/docs/invariants.md b/docs/invariants.md index c135b8b56..1e5c4f8c3 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -82,7 +82,7 @@ not head/idx. | Parent create_fk | **same-batch** planned fks (offline at pin) → **in-flight** (lookup snapshots `drain_and_fence_hi` **before** the wave's TipOnly read and passes it on the last load batch; load drops tagged map rows with pack height **below** that snapshot after that batch's in-flight read; equality keeps; not Class C tip, not `class_a_hi`, not write freeze; one load-thread HashMap, insert after stamp) → **skeleton** (`BatchParentIds` on the `LoadBatch`: lookup TipOnly fk + body_range + spent_range + per-chunk need-vouts) → **Corrupt** on IBD miss. plan=None / S0 (`skeleton = None`) is in-flight → leftover TipOnly. One helper: [`stamp_external_parents`](../crates/rbitcoin-query/src/stamp.rs). No leftover pending map, no process pin FIFO, no BQ-side hits map, no parent-store create_fk on stamp, no published live_union. Same-wave creates are omitted from TipOnly need. Header-cache GC polls store tip each load pack. One fk per txid — [`errata.md`](./errata.md). | Miss of in-flight and skeleton → `Corrupt("parent create_fk unresolved")` (**engine fault**: requeue once, then halt IBD; never blacklist). Identity without loc range → `Corrupt("invariant: loc range missing after identity")`, not a miss | | io_uring harvest | TLS session fail-closed ([`io-modality.md`](./io-modality.md)) | **No** silent success. `Corrupt("invariant: io_uring …")` (not `bdz g page bad slot`). Ring-unavailable still pread-fallback | | Load body outs | By `txout` range only from lookup stamp; incomplete outs → hard Err. Pin **copies** lookup `spent_range` (no loc IO) | **No** loc cold outs on load; **no** `create.loc` on load; **no** `inwit` on pin | -| Ensure (write) | Every non-null spend edge has `spent_range` abs after ensure returns. Lookup stamped archived parents; same-batch from append RAM loc; just-written from write loc packs kept until write of `lookup_started_hi` at note (last TipOnly that may have missed disk loc). Missing abs → `invariant:` | **No** write `create.loc` read | +| Ensure (write) | Every non-null spend edge has `spent_range` abs after ensure returns. Lookup stamped archived parents; same-batch from append RAM loc; just-written from write-thread loc packs kept until write of `lookup_started_hi` at note (extended while the pack is still at/above the load drain fence so InFlight-skip TipOnly can still fill). Missing abs → `invariant:` | **No** write `create.loc` read | | Structural spentness | Abs required for every non-null spend create_fk after load; multi-list → confirmed-strong walk (reorg protocol) | **No** unpinned “wire-corrected create_fk” soft spentness. Multi flag alone is **not** hard `Err` | | Pin create identity | Pin must carry non-zero create txid from **lookup stamp** (plan reverse map / wire prev_txid / `txid.body`) | Soft zero-identity pin → assemble mismatch → cold recovery is **forbidden** | | Tip already-archived | `plan=None`: lookup still stamps parent pin material; load `txout` by range | Soft spentness recovery for zero pin identity is **not** OK | @@ -162,7 +162,7 @@ packs at/above the leaving **pack** height **before** the next bind. | `pin_for_wire_incomplete_outs_is_invariant_error` | `pin_for_wire_batch` incomplete outs → cold miss | | `post_commit_missing_denserels_is_invariant_error` | `post_commit` abs-only annotate | | `ensure_spend_abs_incomplete_is_invariant_error` | `ensure_spend_abs_layouts` post-condition | -| `write_ensure_stamps_spent_range_after_load_pin` / `pin_and_ensure_journey` / `fill_same_batch_abs_from_append_loc_ram` | load pin copies lookup spent range; missing stamp is Corrupt; same-batch abs from append RAM; just-written loc until write of `lookup_started_hi` (no write loc pread) | +| `write_ensure_stamps_spent_range_after_load_pin` / `pin_and_ensure_journey` / `fill_same_batch_abs_from_append_loc_ram` | load pin copies lookup spent range; missing stamp is Corrupt; same-batch abs from append RAM; just-written loc until overlapping lookup write (pack below drain fence; no write loc pread) | | `fill_missing_parent_ranges_stamps_spent_idx_for_archived` | lookup stamp carries spent range for TipOnly leftover | | `spend_abs_jobs_unique_and_missing_is_corrupt` | pin arithmetic abs list; missing → Corrupt | | `structural_pinned_without_abs_is_invariant_error` | `structural_validate_spends` pin without denserels |