diff --git a/AGENTS.md b/AGENTS.md index b2e06427..da236640 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,16 +90,13 @@ Pin material is **plan / batch only** (`batch_pin`, `BatchParents`). No process create pin FIFO. IBD confirm intake is **body queue wire only** → lookup → load. -**RecentCreates** is a write-published Arc layer chain (same splice as -live_union): identity **and** `CreatePin` outs after Class A+idx. Each layer -`until = lookup_started_hi.max(hi)` at publish; drop when `class_a_hi >= until`. -Not BQ depth, not EWMA. Stamp attaches the CreatePin; pin does not re-walk the -ring. Not a coins cache or process pin FIFO. +In-flight CreatePin layers drop at drain+fence **and**, for pin layers, write +`until = lookup_started_hi.max(hi)` then `class_a_hi >= until`. Stamp attaches +the CreatePin from `InFlightView`. Not a coins cache or process pin FIFO. Leftover union, stage IO, S0–S4: **[`docs/invariants.md`](docs/invariants.md)** (the only Allowed/Forbidden IO table). In-flight prune after pin + scripts -handoff; no leftover pending / pin FIFO; RecentCreates drops when Class A -covers lookup-started. Union miss is permanent. +handoff; no leftover pending / pin FIFO. Union miss is permanent. ### Confirm pipeline timers diff --git a/CHANGELOG.md b/CHANGELOG.md index 3aa9a56e..e4b6d8fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,12 @@ before 1.0). ### Changed +- **In-flight keep-until:** pin layers stay until drain+fence **and** + `class_a_hi >= until` (`until = lookup_started_hi.max(hi)` frozen at + write). Stamp walks `InFlightView` only (then live_union, then TipOnly). + [`docs/invariants.md`](docs/invariants.md), + [`docs/concurrency.md`](docs/concurrency.md). + - **`header.head` open-grow replaces the file:** an undersized single-gen OA is deleted and recreated at the create target (`.mlt` kept). Crash after unlink is a missing `header.head`, not a zeroed live table. @@ -169,6 +175,11 @@ before 1.0). ### Removed +- **RecentCreates ring and `PipelineParentStore`:** write no longer clones + a second identity+outs layer list; IBD never used the Weak pin registry. + `ibd: sizes` `recent=` / `pstore=` stay 0. CreatePin outs live on + in-flight keep-until and batch-local `BatchParents`. + - **Dead production APIs:** SH catalog materialize is always k-way (no fan-in reduce / CHECKPOINT / READY, no `RBITCOIN_SH_MERGE_FANIN` / `TARGET_RUN_BYTES` / `MAX_DIRECT_MERGE`). One catalog write policy (no @@ -180,7 +191,7 @@ before 1.0). Unused `NodeClock::mock_value`, `ChainParams::min_difficulty_target`, `outbound_for_ibd`, `InvalidHashSet::{mark_path,is_invalid_fn}`, `MempoolHub::mining_frontier_snapshot`, `ConfirmParentCache::get_header_plan_arc`, - `NodeConfig::with_datadir_cold`. RecentCreates identity **and** outs stay. + `NodeConfig::with_datadir_cold`. ### Fixed diff --git a/OPERATOR.md b/OPERATOR.md index 5a74cff8..7ffb6447 100644 --- a/OPERATOR.md +++ b/OPERATOR.md @@ -228,7 +228,7 @@ onto loadq). Tip-batch getdata races up to 4 peers half-median outlier only after ~60s warm-up and only when the peer pack is not tight (max/min bps > 2×); good-but-slightly-slower peers are kept. -**Create pins:** pipeline-local only (`batch_pin` / `BatchParents`). No process pin FIFO on IBD (no `PipelineParentStore`). Header plans via ConfirmParentCache. Just-confirmed **identity + full create outs** live in RecentCreates Arc layers (drop when Class A covers lookup-started). Not a coins cache. +**Create pins:** pipeline-local only (`batch_pin` / `BatchParents`). No process pin FIFO. Header plans via ConfirmParentCache. Just-confirmed **identity + full create outs** stay on in-flight until write keep-until (`class_a_hi >= lookup_started_hi.max(hi)`). Not a coins cache. **Archive `tx.head` split (perf_dbg):** `plan_batch … head_rd=` is parent **read** resolve (`get_fk_by_txid_batch`, with `probe` / `idx` / `body` subtimers). diff --git a/crates/rbitcoin-consensus/src/confirm_run/bq_resolve.rs b/crates/rbitcoin-consensus/src/confirm_run/bq_resolve.rs index bc0b8e16..fcbee1f1 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/bq_resolve.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/bq_resolve.rs @@ -1057,7 +1057,6 @@ mod tests { &[g_cb.to_byte_array()], &rbitcoin_query::InFlightView::empty(), published.as_ref(), - q.recent_creates().as_ref(), ) .expect("stamp helper after wave"); assert_eq!( @@ -1198,7 +1197,6 @@ mod tests { parent_hash: None, next_tx_start: q.tx_body_count().saturating_add(1).max(1), in_flight: view, - parent_store: None, published: std::sync::Arc::new(rbitcoin_query::PublishedIds::new()), }; let items = [(Height(1), std::sync::Arc::new(b1), None)]; @@ -1292,7 +1290,6 @@ mod tests { parent_hash: None, next_tx_start: q.tx_body_count().saturating_add(1).max(1), in_flight: log.snapshot(), - parent_store: None, published: std::sync::Arc::new(rbitcoin_query::PublishedIds::new()), }; let items = [(Height(1), std::sync::Arc::new(b1), None)]; diff --git a/crates/rbitcoin-consensus/src/confirm_run/lookup.rs b/crates/rbitcoin-consensus/src/confirm_run/lookup.rs index 5de7165c..e7d0d4ba 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/lookup.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/lookup.rs @@ -7,7 +7,7 @@ use super::*; pub struct DenserelsWarmStats { /// Unique external parent creates considered (stamped create_fk, not same-batch). pub parents: u32, - /// Already covered via in-flight / same-batch / RecentCreates outs / pstore adopt. + /// Already covered via in-flight / same-batch / stamp-carried outs. pub already: u32, /// Cold denserels body loads (`txout` by stamped range). Always 0 on the /// shipped pin path — range-fill is `PIN_NEW`, not this field. @@ -172,7 +172,6 @@ pub(super) fn stamp_parent_pin_archived( &need_vec, ifo, query.published_ids(), - query.recent_creates(), ) .map_err(ConsensusError::from)?; let mut stamp = ParentPinStamp { @@ -235,7 +234,6 @@ pub fn confirm_wire_load_from_plan( } = stamped; let ifo = pipeline.map(|p| &p.in_flight); - let parent_store = pipeline.and_then(|p| p.parent_store.as_ref()); let (batch_parents, spend_edges, _warm) = pin_for_wire_batch( query, plan.as_ref(), @@ -243,7 +241,6 @@ pub fn confirm_wire_load_from_plan( &metas, &wire_blocks, ifo, - parent_store, )?; if let Some(ref mut p) = plan { p.freeze_after_pin(); @@ -850,7 +847,6 @@ mod tests { &[parent_txid], &rbitcoin_query::InFlightView::empty(), q.published_ids(), - q.recent_creates(), ) .expect("shared helper"); diff --git a/crates/rbitcoin-consensus/src/confirm_run/mod.rs b/crates/rbitcoin-consensus/src/confirm_run/mod.rs index c074f566..717bc5a1 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/mod.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/mod.rs @@ -83,10 +83,7 @@ pub use scripts::{ }; pub use write::confirm_write_phase; #[cfg(test)] -use write::{ - recent_create_height_slices, recent_create_rows_for_slices, write_batch_vs_tip, - write_height_needed, WriteBatchVsTip, -}; +use write::{write_batch_vs_tip, write_height_needed, WriteBatchVsTip}; /// Pure-write annotate backend from global `RBITCOIN_IO`. #[inline] @@ -156,11 +153,6 @@ pub struct WireLoadPipeline { /// Load looks up create fk / full CreatePin for parents still only in the /// pipeline (body-ahead-of-head). Built via [`rbitcoin_query::InFlightLog::snapshot`]. pub in_flight: rbitcoin_query::InFlightView, - /// Pipeline-wide sparse parent pin store (Weak map; load get-or-insert only). - /// `None` on IBD (RecentCreates outs + in-flight). Tip-follow may still set - /// `Some`. Batches hold `Arc` handles so concurrent stages share one payload - /// per create when a store is present. - pub parent_store: Option>, /// Lookup-published parent identity union (wave hits still live in the BQ window). pub published: std::sync::Arc, } @@ -439,7 +431,6 @@ pub fn confirm_wire_load_phase_pipelined( let ns_filter_plan = t_fp.elapsed().as_nanos() as u64; let inflight = pipeline.map(|p| &p.in_flight); - let parent_store = pipeline.and_then(|p| p.parent_store.as_ref()); let mut parent_pin = match plan.as_mut() { Some(p) => ParentPinStamp::take_from_plan(p), None => stamp_parent_pin_archived(query, params, &metas, &wire_blocks, inflight)?, @@ -451,7 +442,6 @@ pub fn confirm_wire_load_phase_pipelined( &metas, &wire_blocks, inflight, - parent_store, )?; if let Some(ref mut p) = plan { p.freeze_after_pin(); diff --git a/crates/rbitcoin-consensus/src/confirm_run/pin.rs b/crates/rbitcoin-consensus/src/confirm_run/pin.rs index 843c3b95..c66ae9a2 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/pin.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/pin.rs @@ -4,7 +4,7 @@ use super::*; /// Pin parents for wire load: **only spent parents** (sparse outs). /// -/// Sources: plan/in-flight offline denserels → RecentCreates create_pin → +/// Sources: plan/in-flight offline denserels → stamp-carried CreatePin → /// **txout body by range** from [`ParentPinStamp`] (lookup-stamped). Load never /// reads head / `tx.idx` / `txid.body`. Load **copies** lookup-stamped /// `spent_range` onto pins. Write [`ensure_spend_abs_layouts`] is holes-only. @@ -15,7 +15,6 @@ pub(super) fn pin_for_wire_batch( metas: &[BodyMeta], wire_blocks: &[Arc], in_flight: Option<&rbitcoin_query::InFlightView>, - pipeline_parent_store: Option<&std::sync::Arc>, ) -> Result< ( rbitcoin_query::BatchParents, @@ -168,18 +167,7 @@ pub(super) fn pin_for_wire_batch( confirm_load_stats::PIN_RECENT_OUTS_NS.fetch_add(recent_outs_ns, Ordering::Relaxed); } - let mut batch_parents = match pipeline_parent_store { - Some(store) => rbitcoin_query::BatchParents::with_store( - std::sync::Arc::clone(store), - parent_vouts.len(), - ), - None => rbitcoin_query::BatchParents::with_capacity(parent_vouts.len()), - }; - let t_adopt = Instant::now(); - if pipeline_parent_store.is_some() { - batch_parents.adopt_from_store(parent_vouts.keys().copied()); - } - let adopt_ns = t_adopt.elapsed().as_nanos() as u64; + let mut batch_parents = rbitcoin_query::BatchParents::with_capacity(parent_vouts.len()); let thin_ns = t_thin.elapsed().as_nanos() as u64; if thin_ns > 0 { confirm_load_stats::THIN_NS.fetch_add(thin_ns, Ordering::Relaxed); @@ -190,7 +178,7 @@ pub(super) fn pin_for_wire_batch( let t_plan = Instant::now(); for (id, need) in &parent_vouts { let fk = rbitcoin_primitives::Fk(*id); - // Pure adopt hit: refresh meta only when plan/layout material is present + // Same-batch / in-flight pin: refresh meta only when plan/layout material is present // (skip empty refresh_pin_meta — it would reload outs). if !need.is_empty() && batch_parents.pin_covered(fk, need) { if let Some(pin) = plan_by_id.get(id) { @@ -351,10 +339,6 @@ pub(super) fn pin_for_wire_batch( } let contract_ns = t_contract.elapsed().as_nanos() as u64; - let t_publish = Instant::now(); - batch_parents.publish_to_store(); - let publish_ns = t_publish.elapsed().as_nanos() as u64; - let n_unique = parent_vouts.len() as u64; if n_unique > 0 { confirm_load_stats::PARENT_UNIQUE.fetch_add(n_unique, Ordering::Relaxed); @@ -370,25 +354,19 @@ pub(super) fn pin_for_wire_batch( if plan_pin_ns > 0 { confirm_load_stats::PLAN_PIN_NS.fetch_add(plan_pin_ns, Ordering::Relaxed); } - if adopt_ns > 0 { - confirm_load_stats::PIN_ADOPT_NS.fetch_add(adopt_ns, Ordering::Relaxed); - } if contract_ns > 0 { confirm_load_stats::PIN_CONTRACT_NS.fetch_add(contract_ns, Ordering::Relaxed); } - if publish_ns > 0 { - confirm_load_stats::PIN_PUBLISH_NS.fetch_add(publish_ns, Ordering::Relaxed); - } // Last-batch pin residual for slow-load logs (overwrite; not window-summed). let cold_batch_ns = cold_range_batch_ns .saturating_add(cold_io_ns) .saturating_add(cold_decode_ns); confirm_load_stats::note_last_pin( - adopt_ns, + 0, plan_pin_ns, cold_batch_ns, contract_ns, - publish_ns, + 0, n_plan_pin, n_cold.saturating_add(n_range_new), ); diff --git a/crates/rbitcoin-consensus/src/confirm_run/write.rs b/crates/rbitcoin-consensus/src/confirm_run/write.rs index f00d4e1c..ed03b711 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write.rs @@ -91,70 +91,6 @@ pub(super) fn write_batch_vs_tip( } } -/// Per-height ranges over a write batch's `planned_fks` / pins. -/// -/// One RecentCreates fifo row per prepared height. A leftover tail (count -/// mismatch) is tagged with the last height so no create is dropped. -pub(super) fn recent_create_height_slices( - prepared: &[(u32, usize)], - total: usize, -) -> Vec<(u32, std::ops::Range)> { - let mut out = Vec::new(); - let mut off = 0usize; - for &(h, n) in prepared { - if off >= total { - break; - } - if off.saturating_add(n) > total { - break; - } - if n > 0 { - out.push((h, off..off + n)); - } - off = off.saturating_add(n); - } - if off < total { - let height = prepared.last().map(|(h, _)| *h).unwrap_or(0); - out.push((height, off..total)); - } - out -} - -/// Pair idx ranges back to per-height RecentCreates rows (skip missing idx). -pub(super) fn recent_create_rows_for_slices( - slices: &[(u32, std::ops::Range)], - txid_fks: &[([u8; 32], rbitcoin_primitives::Fk)], - ranges: &[Option<(u64, u64)>], - pins: &[rbitcoin_query::CreatePin], -) -> Vec<( - u32, - Vec<( - [u8; 32], - rbitcoin_primitives::Fk, - (u64, u64), - Option, - )>, -)> { - let mut out = Vec::new(); - for (height, range) in slices { - let mut rows = Vec::new(); - for i in range.clone() { - let Some((txid, fk)) = txid_fks.get(i) else { - break; - }; - let Some(body) = ranges.get(i).copied().flatten() else { - continue; - }; - let pin = pins.get(i).map(std::sync::Arc::clone); - rows.push((*txid, *fk, body, pin)); - } - if !rows.is_empty() { - out.push((*height, rows)); - } - } - out -} - /// COMMIT STAGE: optional Class A plan commit → structural → class_c → spend annotate → tip GC /// → optional SP tweak index (**Tip write-through only**; Direct defers to backfill). /// @@ -195,15 +131,18 @@ pub fn confirm_write_phase( fill_packed_ins_from_wire(&mut plan, &batch.prepared, &batch.wire_blocks)?; let t_take = Instant::now(); let planned_fks = plan.planned_fks.clone(); - let pins: Vec = - if plan.batch_pin.len() == plan.planned_fks.len() { + let pins = if query.index_mode().is_tip() { + Some(if plan.batch_pin.len() == plan.planned_fks.len() { std::mem::take(&mut plan.batch_pin) } else { plan.packed .iter() .map(|(pin, _)| std::sync::Arc::clone(pin)) - .collect() - }; + .collect::>() + }) + } else { + None + }; plan_take_ns = t_take.elapsed().as_nanos() as u64; let t_ca = Instant::now(); let committed = query @@ -214,7 +153,7 @@ pub fn confirm_write_phase( // already present) uses store denserels via ensure / class_c cold pins. // Direct SH collect is a no-op — skip the FkMap. if committed { - if query.index_mode().is_tip() { + if let Some(pins) = pins { let t_map = Instant::now(); write_create_pins.reserve(planned_fks.len()); for (fk, pin) in planned_fks.iter().zip(pins.iter()) { @@ -222,13 +161,11 @@ pub fn confirm_write_phase( } create_map_ns = t_map.elapsed().as_nanos() as u64; } - let t_idx = Instant::now(); + let t_ens = Instant::now(); let body_ranges = query .store() .tx_body_range_batch(&planned_fks) .map_err(ConsensusError::from)?; - let idx_ns = t_idx.elapsed().as_nanos() as u64; - let t_ens = Instant::now(); fill_planned_create_layout_after_commit( query, &mut batch.batch_parents, @@ -236,41 +173,11 @@ pub fn confirm_write_phase( &body_ranges, )?; ensure_ns = ensure_ns.saturating_add(t_ens.elapsed().as_nanos() as u64); - let slices = recent_create_height_slices( - &batch - .prepared - .iter() - .map(|p| (p.height.0, p.tx_fks.len())) - .collect::>(), - planned_fks.len(), - ); - let t_recent = Instant::now(); - let txid_fks: Vec<([u8; 32], rbitcoin_primitives::Fk)> = planned_fks - .iter() - .zip(pins.iter()) - .map(|(fk, pin)| (pin.0.txid, *fk)) - .collect(); - for (height, rows) in - recent_create_rows_for_slices(&slices, &txid_fks, &body_ranges, &pins) - { - query.note_recent_creates_pins(height, rows); - } if let Some(last) = batch.prepared.last() { query.set_class_a_hi(Some(last.height.0)); } - let t_clone = Instant::now(); - query.flush_recent_creates(); - let clone_ns = t_clone.elapsed().as_nanos() as u64; - let recent_ns = t_recent.elapsed().as_nanos() as u64; - if idx_ns > 0 { - confirm_phase_stats::WRITE_RECENT_IDX_NS.fetch_add(idx_ns, Ordering::Relaxed); - } - if clone_ns > 0 { - confirm_phase_stats::WRITE_RECENT_CLONE_NS - .fetch_add(clone_ns, Ordering::Relaxed); - } - if recent_ns > 0 { - confirm_phase_stats::WRITE_RECENT_NS.fetch_add(recent_ns, Ordering::Relaxed); + for p in &batch.prepared { + query.stamp_create_keep_until(p.height.0); } } } @@ -395,7 +302,7 @@ pub fn confirm_write_phase( /// After Class A commit, set body_range (+ spent.idx) for **pinned** creates /// still missing layout. Body ranges come from the write's one `tx_body_range_batch` -/// (shared with RecentCreates). Spent holes use one `tx_spent_range_batch`. +/// (same batch as layout fill). Spent holes use one `tx_spent_range_batch`. 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 75d915b1..f0b795a4 100644 --- a/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs +++ b/crates/rbitcoin-consensus/src/confirm_run/write_idempotent_tests.rs @@ -1,8 +1,8 @@ //! Confirm_run unit tests (peeled from confirm_run.rs). use super::{ - confirm_archive_kind, recent_create_height_slices, recent_create_rows_for_slices, - write_batch_vs_tip, write_height_needed, ConfirmArchiveKind, WriteBatchVsTip, + confirm_archive_kind, write_batch_vs_tip, write_height_needed, ConfirmArchiveKind, + WriteBatchVsTip, }; #[test] @@ -17,77 +17,6 @@ fn tx_head_drain_thread_is_named_and_reused() { assert_eq!(id1, id2, "drain must keep one OS thread across batches"); } -#[test] -fn recent_create_height_slices_two_heights_and_remainder() { - assert_eq!( - recent_create_height_slices(&[(10, 2), (11, 3)], 5), - vec![(10, 0..2), (11, 2..5)] - ); - assert_eq!( - recent_create_height_slices(&[(10, 2), (11, 3)], 7), - vec![(10, 0..2), (11, 2..5), (11, 5..7)], - "tail past prepared counts tags the last height" - ); - assert!(recent_create_height_slices(&[(10, 2)], 0).is_empty()); - assert_eq!( - recent_create_height_slices(&[(10, 0), (11, 4)], 4), - vec![(11, 0..4)] - ); -} - -#[test] -fn recent_create_rows_skip_missing_idx_keep_heights() { - let tid = |b| { - let mut t = [0u8; 32]; - t[0] = b; - t - }; - let slices = recent_create_height_slices(&[(10, 2), (11, 2)], 4); - let pairs = [ - (tid(1), rbitcoin_primitives::Fk(1)), - (tid(2), rbitcoin_primitives::Fk(2)), - (tid(3), rbitcoin_primitives::Fk(3)), - (tid(4), rbitcoin_primitives::Fk(4)), - ]; - let ranges = [Some((1, 8)), None, Some((9, 8)), Some((17, 8))]; - let rows = recent_create_rows_for_slices(&slices, &pairs, &ranges, &[]); - assert_eq!(rows.len(), 2); - assert_eq!(rows[0].0, 10); - assert_eq!(rows[0].1.len(), 1, "missing idx at height 10 dropped"); - assert_eq!(rows[1].0, 11); - assert_eq!(rows[1].1.len(), 2); -} - -#[test] -fn recent_create_rows_share_create_pin_slice() { - use rbitcoin_store::{OutputRecord, TxRecord}; - use std::sync::Arc; - let tid = [0x11u8; 32]; - let pin = Arc::new(( - TxRecord { - txid: tid, - version: 1, - locktime: 0, - input_start_fk: rbitcoin_primitives::Fk::NULL, - input_count: 0, - output_start_fk: rbitcoin_primitives::Fk::NULL, - output_count: 1, - }, - vec![OutputRecord::unspent(1, vec![0x51])], - )); - let slices = recent_create_height_slices(&[(10, 1)], 1); - let pairs = [(tid, rbitcoin_primitives::Fk(1))]; - let ranges = [Some((8, 16))]; - let pins = [Arc::clone(&pin)]; - let rows = recent_create_rows_for_slices(&slices, &pairs, &ranges, &pins); - let got = rows[0].1[0].3.as_ref().expect("pin"); - assert!( - Arc::ptr_eq(got, &pin), - "rows must Arc-clone the pin slice, not rebuild outs" - ); - assert_eq!(rows[0].1[0].2, (8, 16)); -} - /// Batch append: contiguous heights merge; gap returns Err(other). #[test] fn script_ok_append_contiguous_and_gap() { @@ -1185,7 +1114,7 @@ fn pin_and_ensure_journey() { plan.planned_fks = vec![Fk(1)]; let mut stamp = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let err = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let err = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect_err("missing parent must hard-fail pin"); let msg = format!("{err}"); assert!( @@ -1253,7 +1182,7 @@ fn pin_and_ensure_journey() { ); let mut stamp = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let (parents, _, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let (parents, _, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect("pin via stamped range"); assert!(parents.contains(pfk)); assert!(parents.get_parent_out(pfk, 0).is_some()); @@ -1275,7 +1204,7 @@ fn pin_and_ensure_journey() { ); let mut empty_stamp = ParentPinStamp::default(); fill_edges_from_packed(&mut plan2); - let err = pin_for_wire_batch(&q, Some(&plan2), &mut empty_stamp, &[], &[], None, None) + let err = pin_for_wire_batch(&q, Some(&plan2), &mut empty_stamp, &[], &[], None) .expect_err("plan maps must not backfill an empty stamp"); assert!(err.to_string().contains("lookup stage miss"), "got: {err}"); @@ -1328,7 +1257,7 @@ fn pin_and_ensure_journey() { let mut stamp3 = ParentPinStamp::take_from_plan(&mut plan3); fill_edges_from_packed(&mut plan3); let (mut parents3, _, _) = - pin_for_wire_batch(&q, Some(&plan3), &mut stamp3, &[], &[], None, None).unwrap(); + pin_for_wire_batch(&q, Some(&plan3), &mut stamp3, &[], &[], None).unwrap(); assert!( parents3.has_abs_layout(pfk), "load pin copies lookup-stamped spent.idx range (no write idx)" @@ -1362,7 +1291,7 @@ fn pin_and_ensure_journey() { let mut stamp4 = ParentPinStamp::take_from_plan(&mut plan4); fill_edges_from_packed(&mut plan4); let (parents4, _, _) = - pin_for_wire_batch(&q, Some(&plan4), &mut stamp4, &[], &[], None, None).unwrap(); + pin_for_wire_batch(&q, Some(&plan4), &mut stamp4, &[], &[], None).unwrap(); assert!( !parents4.contains(Fk(2)), "same-header create is wire-valued, not pinned" @@ -1371,126 +1300,6 @@ fn pin_and_ensure_journey() { let _ = std::fs::remove_dir_all(&path); } -/// Cold-range pin then pstore adopt: first pin reads body, second does not. -#[test] -fn pin_for_wire_cold_range_then_adopt_skips_body_io() { - use super::{pin_for_wire_batch, ParentPinStamp}; - use rbitcoin_primitives::Fk; - use rbitcoin_query::{PipelineParentStore, Query}; - use rbitcoin_store::{InputRecord, OutputRecord, TxRecord}; - use std::sync::{Arc, Once}; - - static ONCE: Once = Once::new(); - ONCE.call_once(|| { - if std::env::var_os("RBITCOIN_HEAD_SCALE").is_none() { - std::env::set_var("RBITCOIN_HEAD_SCALE", "tiny"); - } - }); - let path = std::env::temp_dir().join(format!( - "rbitcoin-pin-range-adopt-{}-{}", - std::process::id(), - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_nanos()) - .unwrap_or(0) - )); - let _ = std::fs::remove_dir_all(&path); - let q = Query::open_or_create(&path).unwrap(); - - let parent_tx = TxRecord { - txid: [0xab; 32], - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 1, - output_start_fk: Fk::NULL, - output_count: 1, - }; - let parent_outs = vec![OutputRecord::unspent(50_0000_0000, vec![0x51])]; - let parent_ins = vec![InputRecord::coinbase(u32::MAX, vec![0x01], vec![])]; - let pfk = q - .store() - .txs - .put_full_batch_indexed(&[(parent_tx.clone(), parent_ins, parent_outs)], true) - .unwrap()[0]; - let range = q.store().tx_body_range(pfk).unwrap(); - - let spend_tx = TxRecord { - txid: [0xcd; 32], - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 1, - output_start_fk: Fk::NULL, - output_count: 1, - }; - let spend_outs = vec![OutputRecord::unspent(1, vec![0x51])]; - let spend_ins = vec![InputRecord { - prev_txid: parent_tx.txid, - create_fk: pfk, - prev_index: 0, - sequence: u32::MAX, - script_sig: vec![], - witness: vec![], - }]; - let stamp_plan = || { - let mut plan = rbitcoin_query::ArchiveWritePlan::empty(); - plan.packed = vec![( - std::sync::Arc::new((spend_tx.clone(), spend_outs.clone())), - spend_ins.clone(), - )]; - plan.planned_fks = vec![Fk(2)]; - if let Some(id) = pfk.get() { - plan.external_parents.insert( - id, - rbitcoin_query::ParentIdent::with_body(parent_tx.txid, range), - ); - } - plan - }; - - let store = Arc::new(PipelineParentStore::new()); - let mut plan = stamp_plan(); - let mut parent_pin = ParentPinStamp::take_from_plan(&mut plan); - fill_edges_from_packed(&mut plan); - let (parents, _thin, _warm) = pin_for_wire_batch( - &q, - Some(&plan), - &mut parent_pin, - &[], - &[], - None, - Some(&store), - ) - .unwrap(); - assert!(parents.contains(pfk)); - assert!( - parents.get_parent_out(pfk, 0).is_some(), - "cold-range pin must load the spent vout" - ); - - let mut plan2 = stamp_plan(); - let mut parent_pin2 = ParentPinStamp::take_from_plan(&mut plan2); - fill_edges_from_packed(&mut plan2); - let (parents2, _thin2, _warm2) = pin_for_wire_batch( - &q, - Some(&plan2), - &mut parent_pin2, - &[], - &[], - None, - Some(&store), - ) - .unwrap(); - assert!(parents2.contains(pfk)); - assert!( - parents2.get_parent_out(pfk, 0).is_some(), - "pstore adopt must still serve the spent vout" - ); - - let _ = std::fs::remove_dir_all(&path); -} - /// Wire pin: in-flight outs shorter than need → cold miss → hard invariant. #[test] fn pin_for_wire_incomplete_outs_is_invariant_error() { @@ -1575,7 +1384,7 @@ fn pin_for_wire_incomplete_outs_is_invariant_error() { let mut parent_pin = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let err = pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], Some(&ifo), None) + let err = pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], Some(&ifo)) .expect_err("incomplete outs must hard-fail pin"); let msg = format!("{err}"); assert!( @@ -1708,7 +1517,7 @@ fn pin_takes_stamp_parent_vouts() { Some(&[0u32][..]) ); fill_edges_from_packed(&mut plan); - let (parents, _, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let (parents, _, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect("pin via taken vouts"); assert!(stamp.parent_vouts.is_empty(), "pin must take stamp vouts"); assert!(parents.contains(pfk)); @@ -1786,7 +1595,7 @@ fn pin_for_wire_create_pin_shares_script_bytes() { plan.external_parent_vouts.insert(1, vec![0]); let mut stamp = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let (parents, edges, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let (parents, edges, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect("cross-height CreatePin pin"); let child_edges = edges.get(&2).expect("child spend edges"); assert_eq!(child_edges.len(), 1); @@ -1879,7 +1688,7 @@ fn pin_plan_edges_without_packed_ins() { ); let mut stamp = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let (parents, edges, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let (parents, edges, _) = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect("pin from plan.edges with empty packed ins"); let child_edges = edges.get(&2).expect("child spend edges"); assert_eq!(child_edges.len(), 1); @@ -1932,7 +1741,7 @@ fn pin_plan_empty_edges_is_invariant() { plan.planned_fks = vec![Fk(1)]; plan.batch_pin = vec![Arc::clone(&pin)]; let mut stamp = ParentPinStamp::take_from_plan(&mut plan); - let err = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None, None) + let err = pin_for_wire_batch(&q, Some(&plan), &mut stamp, &[], &[], None) .expect_err("empty edges with planned fks must not skip spends"); let msg = format!("{err}"); assert!( @@ -2034,7 +1843,7 @@ fn pin_sparse_need_high_vout_only() { let mut parent_pin = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); let (parents, _thin, _warm) = - pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None, None) + pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None) .expect("pin high vout"); assert!(parents.get_parent_out(pfk, 3).is_some()); assert_eq!( @@ -2050,14 +1859,11 @@ fn pin_sparse_need_high_vout_only() { } /// Range-fill this window is `PIN_NEW`, not `PIN_CACHE_BODY` / `warm.already`. -/// -/// Adopt 1 + cold-range 2 → `already=1` (cache), not 3. `pin_hit%` is -/// `1/(1+2)=33`, not “we just loaded them.” #[test] fn pin_range_fill_does_not_count_as_cache_hit() { use super::{pin_for_wire_batch, ParentPinStamp}; use rbitcoin_primitives::Fk; - use rbitcoin_query::{ArchiveWritePlan, BatchParents, PipelineParentStore, Query}; + use rbitcoin_query::{ArchiveWritePlan, Query}; use rbitcoin_store::{InputRecord, OutputRecord, TxRecord}; use std::sync::{Arc, Once}; @@ -2105,20 +1911,6 @@ fn pin_range_fill_does_not_count_as_cache_hit() { ranges.push(q.store().tx_body_range(*fk).unwrap()); } - // Live pin for parent 0 only (same Weak lifecycle as outs share). - let store = Arc::new(PipelineParentStore::new()); - let mut keep = BatchParents::with_store(Arc::clone(&store), 1); - keep.insert_owned( - fks[0], - items[0].0.clone(), - vec![(0, items[0].2[0].clone())], - vec![0], - Some(false), - Some(ranges[0]), - Vec::new(), - ); - keep.publish_to_store(); - let spend_tx = TxRecord { txid: [0x5cu8; 32], version: 1, @@ -2155,29 +1947,20 @@ fn pin_range_fill_does_not_count_as_cache_hit() { let mut parent_pin = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); - let (_parents, _thin, warm) = pin_for_wire_batch( - &q, - Some(&plan), - &mut parent_pin, - &[], - &[], - None, - Some(&store), - ) - .expect("adopt 1 + range-fill 2"); + let (_parents, _thin, warm) = + pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None).expect("range-fill 3"); assert_eq!(warm.parents, 3); assert_eq!( - warm.already, 1, + warm.already, 0, "range-fills must not increment already / PIN_CACHE_BODY" ); - drop(keep); let _ = std::fs::remove_dir_all(&path); } -/// Write-published RecentCreates outs cover a later spend after in-flight is gone. -/// That is `PIN_CACHE_BODY` / `warm.already`, not `PIN_NEW` / range-fill. +/// Stamp-carried CreatePin outs cover a later spend. That is `PIN_CACHE_BODY` +/// / `warm.already`, not `PIN_NEW` / range-fill. #[test] -fn pin_recent_outs_is_cache_not_new() { +fn pin_stamp_outs_is_cache_not_new() { use super::{pin_for_wire_batch, ParentPinStamp}; use rbitcoin_primitives::Fk; use rbitcoin_query::{ArchiveWritePlan, CreatePin, Query}; @@ -2215,8 +1998,6 @@ fn pin_recent_outs_is_cache_not_new() { let parent_out = OutputRecord::unspent(50, vec![0x51, 0xaa]); let pin: CreatePin = Arc::new((parent_tx.clone(), vec![parent_out.clone()])); let pfk = Fk(7); - q.note_recent_creates_pins(10, [(tid, pfk, (1, 8), Some(Arc::clone(&pin)))]); - q.flush_recent_creates(); let spend_tx = TxRecord { txid: [0x5cu8; 32], @@ -2256,20 +2037,19 @@ fn pin_recent_outs_is_cache_not_new() { Arc::ptr_eq(parent_pin.create_pin(7).expect("stamp pin"), &pin), "pin must use stamp-carried CreatePin" ); - q.recent_creates().drop_from(0); fill_edges_from_packed(&mut plan); let (_parents, _thin, warm) = - pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None, None) - .expect("stamp-carried outs must cover without a live RecentCreates ring"); + pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None) + .expect("stamp-carried outs must cover"); assert_eq!(warm.parents, 1); assert_eq!( warm.already, 1, - "RecentCreates outs must count as PIN_CACHE, not PIN_NEW" + "stamp-carried outs must count as PIN_CACHE, not PIN_NEW" ); let _ = std::fs::remove_dir_all(&path); } -/// Identity-only RecentCreates (no outs) still cold-fills by stamped range. +/// Identity-only stamp (no outs) still cold-fills by stamped range. #[test] fn pin_recent_identity_without_outs_still_range_fills() { use super::{pin_for_wire_batch, ParentPinStamp}; @@ -2317,8 +2097,6 @@ fn pin_recent_identity_without_outs_still_range_fills() { .put_full_batch_indexed(&[parent.clone()], true) .unwrap(); let range = q.store().tx_body_range(fks[0]).unwrap(); - q.note_recent_creates_rows(10, [(tid, fks[0], range)]); - q.flush_recent_creates(); let spend_tx = TxRecord { txid: [0x5du8; 32], @@ -2351,8 +2129,8 @@ fn pin_recent_identity_without_outs_still_range_fills() { let mut parent_pin = ParentPinStamp::take_from_plan(&mut plan); fill_edges_from_packed(&mut plan); let (_parents, _thin, warm) = - pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None, None) - .expect("identity-only recent still range-fills"); + pin_for_wire_batch(&q, Some(&plan), &mut parent_pin, &[], &[], None) + .expect("identity-only stamp still range-fills"); assert_eq!(warm.parents, 1); assert_eq!( warm.already, 0, @@ -2717,9 +2495,9 @@ fn structural_pinned_without_abs_is_invariant_error() { let _ = std::fs::remove_dir_all(&path); } -/// Direct write skips SH FkMap; RecentCreates body ranges match idx. +/// Direct write skips SH FkMap; Class A idx holds the body range. #[test] -fn direct_write_skips_create_pin_map_recent_matches_idx() { +fn direct_write_skips_create_pin_map_idx_without_recent() { use crate::regtest_pad::mine_empty_regtest; use crate::{accept_and_connect_block, ChainParams, Milestone}; use bitcoin::hashes::Hash; @@ -2751,11 +2529,11 @@ fn direct_write_skips_create_pin_map_recent_matches_idx() { let b1 = mine_empty_regtest(genesis.block_hash(), genesis.header.time + 600, 1); let tid = b1.txdata[0].compute_txid().to_byte_array(); accept_and_connect_block(&q, ¶ms, Height(1), &b1, Milestone::NONE).unwrap(); - let (fk, range) = q - .recent_creates() - .get(&tid) - .expect("height-1 create stays in RecentCreates while lookup_started_hi is ahead"); + let fk = q + .tx_fk_by_txid(&tid) + .expect("txid lookup") + .expect("height-1 create on idx"); let idx = q.store().tx_body_range(fk).expect("idx after Class A"); - assert_eq!(range, idx, "RecentCreates body range must be the idx row"); + assert!(idx.1 > 0, "Class A body range must be on idx"); let _ = std::fs::remove_dir_all(&path); } diff --git a/crates/rbitcoin-net/src/chain.rs b/crates/rbitcoin-net/src/chain.rs index 33f08924..a66354c4 100644 --- a/crates/rbitcoin-net/src/chain.rs +++ b/crates/rbitcoin-net/src/chain.rs @@ -2547,9 +2547,6 @@ mod tests { parent_hash: None, next_tx_start: hub.query.tx_body_count().saturating_add(1).max(1), in_flight: rbitcoin_query::InFlightView::empty(), - parent_store: Some(std::sync::Arc::new( - rbitcoin_query::PipelineParentStore::new(), - )), published: std::sync::Arc::new(rbitcoin_query::PublishedIds::new()), }; let mat1 = hub diff --git a/crates/rbitcoin-net/src/ibd/confirm/mod.rs b/crates/rbitcoin-net/src/ibd/confirm/mod.rs index cabde5f6..c1d82d5c 100644 --- a/crates/rbitcoin-net/src/ibd/confirm/mod.rs +++ b/crates/rbitcoin-net/src/ibd/confirm/mod.rs @@ -59,9 +59,12 @@ impl LoadAheadState { /// `next_tx_start` still tracks body count (next free create fk). fn prune_committed(&mut self, hub: &ChainHub) { let body_n = hub.query.tx_body_count(); + self.in_flight + .apply_keep_untils(hub.query.take_create_keep_until()); self.in_flight.prune_if_head_ready( &hub.query.store().height_fence_snapshot(), hub.query.head_drain_fk(), + hub.query.class_a_hi(), ); self.next_tx_start = self.next_tx_start.max(body_n.saturating_add(1).max(1)); if let Some((h, _)) = self.last_loaded { @@ -92,7 +95,6 @@ impl LoadAheadState { parent_hash, next_tx_start: self.next_tx_start, in_flight: self.in_flight.snapshot(), - parent_store: None, published: std::sync::Arc::clone(&self.published), } } diff --git a/crates/rbitcoin-query/src/archive.rs b/crates/rbitcoin-query/src/archive.rs index 7f6808e1..76be61b6 100644 --- a/crates/rbitcoin-query/src/archive.rs +++ b/crates/rbitcoin-query/src/archive.rs @@ -660,14 +660,12 @@ impl Query { let need_vec: Vec<[u8; 32]> = need_external.iter().copied().collect(); let collect_ns = t_collect.elapsed().as_nanos() as u64; - // External parents: in-flight → published live_union → recent creates - // → leftover TipOnly, then idx range-fill. Same helper as plan=None. + // External parents: in-flight → published live_union → leftover TipOnly let ext = crate::stamp_external_parents( &self.store, &need_vec, in_flight, published.unwrap_or(self.published_ids.as_ref()), - self.recent_creates.as_ref(), )?; let inflight_ns = ext.inflight_ns; let head_fk_ns = ext.head_fk_ns; @@ -1409,7 +1407,11 @@ mod tests { )); q.archive_commit_plan(plan_a).unwrap(); assert_eq!(q.store().tx_height_get(parent_fk).unwrap(), None); - log.prune_if_head_ready(&q.store().height_fence_snapshot(), q.head_drain_fk()); + log.prune_if_head_ready( + &q.store().height_fence_snapshot(), + q.head_drain_fk(), + q.class_a_hi(), + ); assert!( log.snapshot().get_create_fk(&parent_txid).is_some(), "fence missing: prune must keep" @@ -1491,7 +1493,11 @@ mod tests { q.store() .height_fence_extend(rbitcoin_primitives::Height(0), header_fk) .unwrap(); - log.prune_if_head_ready(&q.store().height_fence_snapshot(), q.head_drain_fk()); + log.prune_if_head_ready( + &q.store().height_fence_snapshot(), + q.head_drain_fk(), + q.class_a_hi(), + ); assert!( log.snapshot().get_create_fk(&parent_txid).is_some(), "drain_fk 0: prune must keep" @@ -1835,7 +1841,6 @@ mod tests { &[parent_txid], &crate::InFlightView::empty(), q.published_ids().as_ref(), - q.recent_creates().as_ref(), ) .expect("stamp archived parent"); assert_eq!(helper.resolved.get(&parent_txid), Some(&Fk(1))); @@ -2025,81 +2030,6 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } - /// Live pipeline pin is outs-only: stamp does not use `bulk_lookup_txid`. - #[test] - fn archive_plan_batch_from_store_pstore_is_not_stamp_source() { - use crate::{BatchParents, PipelineParentStore}; - use std::sync::Arc; - let (dir, q) = temp_query("pin-txid-stamp"); - let parent_txid = { - let mut t = [0u8; 32]; - t[0] = 0x11; - t - }; - let store = Arc::new(PipelineParentStore::new()); - let mut bp = BatchParents::with_store(Arc::clone(&store), 1); - bp.insert_owned( - Fk(99), - TxRecord { - txid: parent_txid, - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 1, - output_start_fk: Fk::NULL, - output_count: 1, - }, - vec![(0, OutputRecord::unspent(1, vec![0x51]))], - vec![0], - Some(false), - Some((5000, 40)), - Vec::new(), - ); - bp.publish_to_store(); - let _keep = bp; - - let child_txid = { - let mut t = [0u8; 32]; - t[0] = 0x22; - t - }; - let child = TxApply { - tx: TxRecord { - txid: child_txid, - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 1, - output_start_fk: Fk::NULL, - output_count: 1, - }, - inputs: vec![InputRecord { - prev_txid: parent_txid, - create_fk: Fk::NULL, - prev_index: 0, - sequence: u32::MAX, - script_sig: vec![], - witness: vec![], - }], - outputs: vec![OutputRecord::unspent(1, vec![0x51])], - }; - let mut need = vec![(Fk(1), vec![child])]; - crate::archive_phase_stats::with_exclusive(|| { - let _ = crate::archive_phase_stats::sample_and_reset(); - let err = q - .archive_plan_batch_from_store(&mut need, 1, &crate::InFlightView::empty(), None) - .expect_err("pstore pin is not a stamp source"); - assert!( - err.to_string().contains("parent create_fk unresolved"), - "got: {err}" - ); - let mix = crate::archive_phase_stats::sample_and_reset(); - assert_eq!(mix.pin_txid_n, 0); - assert!(mix.head_need > 0, "pstore-only parent must leftover"); - }); - let _ = std::fs::remove_dir_all(&dir); - } - /// BQ-ahead facts live on the published layer. A leftover hits map is not /// a stamp source (shipped IBD already passes `None`). #[test] @@ -2190,7 +2120,6 @@ mod tests { &[parent_txid], &crate::InFlightView::empty(), published.as_ref(), - q.recent_creates().as_ref(), ) .expect("shared helper"); assert_eq!(helper.resolved.get(&parent_txid), Some(&Fk(88))); @@ -2217,66 +2146,11 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } - /// Write-published recent identity skips leftover TipOnly (`head_need=0`). - #[test] - fn stamp_hits_recent_creates_skips_leftover() { - let (dir, q) = temp_query("recent-creates-stamp"); - let parent_txid = { - let mut t = [0u8; 32]; - t[0] = 0x91; - t - }; - q.recent_creates() - .note(10, [(parent_txid, Fk(91), (5000, 16))]); - crate::archive_phase_stats::with_exclusive(|| { - let _ = crate::archive_phase_stats::sample_and_reset(); - let helper = crate::stamp_external_parents( - q.store(), - &[parent_txid], - &crate::InFlightView::empty(), - q.published_ids().as_ref(), - q.recent_creates().as_ref(), - ) - .expect("recent stamp"); - assert_eq!(helper.resolved.get(&parent_txid), Some(&Fk(91))); - assert_eq!( - helper.idents.get(&91).and_then(|p| p.body), - Some((5000, 16)) - ); - assert_eq!(helper.recent_n, 1); - assert_eq!(helper.head_need_n, 0, "recent hit must skip leftover"); - assert!( - helper - .idents - .get(&91) - .and_then(|p| p.pin.as_ref()) - .is_none(), - "identity-only recent note must not carry a CreatePin" - ); - - let child = child_spend(parent_txid, 0x92); - let mut need = vec![(Fk(1), vec![child])]; - let plan = q - .archive_plan_batch_from_store(&mut need, 1, &crate::InFlightView::empty(), None) - .expect("S0 recent"); - assert_eq!(plan.packed[0].1[0].create_fk, Fk(91)); - assert_eq!( - plan.external_parents.get(&91).and_then(|p| p.body), - Some((5000, 16)) - ); - let mix = crate::archive_phase_stats::sample_and_reset(); - assert_eq!(mix.head_need, 0, "plan path must skip leftover too"); - assert!(mix.recent_n >= 1, "recent hits must be metered: {mix:?}"); - }); - q.recent_creates().drop_from(10); - assert!(q.recent_creates().get(&parent_txid).is_none()); - let _ = std::fs::remove_dir_all(&dir); - } - + /// In-flight CreatePin skips leftover TipOnly (`head_need=0`) and stamps the pin. #[test] - fn stamp_recent_hit_carries_create_pin() { + fn stamp_inflight_hit_carries_create_pin() { use std::sync::Arc; - let (dir, q) = temp_query("recent-creates-pin"); + let (dir, q) = temp_query("inflight-creates-pin"); let parent_txid = { let mut t = [0u8; 32]; t[0] = 0x93; @@ -2294,31 +2168,42 @@ mod tests { }, vec![rbitcoin_store::OutputRecord::unspent(1, vec![0x51])], )); - q.recent_creates().note_pins( - 10, - [(parent_txid, Fk(93), (5000, 16), Some(Arc::clone(&pin)))], - ); - let helper = crate::stamp_external_parents( - q.store(), - &[parent_txid], - &crate::InFlightView::empty(), - q.published_ids().as_ref(), - q.recent_creates().as_ref(), - ) - .expect("recent stamp"); - let got = helper - .idents - .get(&93) - .and_then(|p| p.pin.as_ref()) - .expect("stamp must carry CreatePin"); - assert!(Arc::ptr_eq(got, &pin)); + let mut log = crate::InFlightLog::new(); + log.note_layer(crate::InFlightLayer::from_plan_pins([(Fk(93), &pin)])); + let ifo = log.snapshot(); + crate::archive_phase_stats::with_exclusive(|| { + let _ = crate::archive_phase_stats::sample_and_reset(); + let helper = crate::stamp_external_parents( + q.store(), + &[parent_txid], + &ifo, + q.published_ids().as_ref(), + ) + .expect("inflight stamp"); + assert_eq!(helper.head_need_n, 0, "inflight hit must skip leftover"); + let got = helper + .idents + .get(&93) + .and_then(|p| p.pin.as_ref()) + .expect("stamp must carry CreatePin"); + assert!(Arc::ptr_eq(got, &pin)); + + let child = child_spend(parent_txid, 0x94); + let mut need = vec![(Fk(1), vec![child])]; + let plan = q + .archive_plan_batch_from_store(&mut need, 1, &ifo, None) + .expect("S0 inflight"); + assert_eq!(plan.packed[0].1[0].create_fk, Fk(93)); + let mix = crate::archive_phase_stats::sample_and_reset(); + assert_eq!(mix.head_need, 0, "plan path must skip leftover too"); + }); let _ = std::fs::remove_dir_all(&dir); } - /// After Class A + idx, `publish_recent_creates` is what write uses. + /// After drain+fence, TipOnly stamps without a RAM identity ring. #[test] - fn publish_recent_creates_after_commit_skips_leftover() { - let (dir, q) = temp_query("recent-publish"); + fn leftover_tiponly_after_commit_skips_when_connected() { + let (dir, q) = temp_query("tiponly-after-commit"); let mut need_a = vec![(Fk(1), vec![coinbase_apply(1)])]; let empty = crate::InFlightView::empty(); let plan_a = q @@ -2327,27 +2212,19 @@ mod tests { let parent_txid = plan_a.batch_creates[0].0; let parent_fk = plan_a.batch_creates[0].1; let header_fk = plan_a.per_header_ranges[0].0; - let creates = plan_a.batch_creates.clone(); q.archive_commit_plan(plan_a).unwrap(); q.store() .height_fence_extend(rbitcoin_primitives::Height(0), header_fk) .unwrap(); - q.publish_recent_creates(0, creates).unwrap(); - assert!( - q.recent_creates().get(&parent_txid).is_some(), - "write publish must expose committed identity" - ); + q.on_load_pack().unwrap(); crate::archive_phase_stats::with_exclusive(|| { let _ = crate::archive_phase_stats::sample_and_reset(); let mut need_b = vec![(Fk(2), vec![child_spend(parent_txid, 0xec)])]; let plan_b = q .archive_plan_batch_from_store(&mut need_b, 2, &empty, None) - .expect("recent publish must stamp"); + .expect("TipOnly must stamp after fence"); assert_eq!(plan_b.packed[0].1[0].create_fk, parent_fk); - let mix = crate::archive_phase_stats::sample_and_reset(); - assert_eq!(mix.head_need, 0, "published recent must skip leftover"); - assert!(mix.recent_n >= 1, "recent publish must meter: {mix:?}"); }); let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-query/src/batch_parents.rs b/crates/rbitcoin-query/src/batch_parents.rs index 45578f39..f4741655 100644 --- a/crates/rbitcoin-query/src/batch_parents.rs +++ b/crates/rbitcoin-query/src/batch_parents.rs @@ -1,20 +1,8 @@ -//! Pipeline-shared sparse parent pins for confirm. +//! Batch-local sparse parent pins for confirm. //! -//! **Sharing:** one [`SharedParentPin`] per create_fk (refcounted via `Arc`) while -//! any in-flight batch needs it. Batches hold a cheap handle map -//! ([`BatchParents`]) of `Arc`s — no deep-copy of outs between stages/batches. -//! -//! **Registry:** [`PipelineParentStore`] keeps `Weak` entries. Prep **does not** -//! take the store mutex on every parent insert (that made free-plan pin ~9× -//! slower). Instead: -//! 1. [`BatchParents::adopt_from_store`] — one lock, upgrade live pins for the -//! batch's parent set (cross-batch RAM share) -//! 2. Local [`BatchParents::insert_owned`] — lock-free HashMap insert (same cost -//! class as the pre-share `ParentEntry` path) -//! 3. [`BatchParents::publish_to_store`] — one lock, publish Weaks / merge races -//! -//! Assemble/write read pin data through the batch's `Arc`s — **no** global map -//! lock on the hot path. +//! **Sharing:** one [`SharedParentPin`] per create_fk (refcounted via `Arc`) +//! while the batch needs it. [`BatchParents`] is a cheap handle map of `Arc`s — +//! no process Weak registry. //! //! **Immutable publish:** outs and layout are **separate** immutable Arc //! snapshots. Vacant `insert_owned` stores Frozen halves (no `ArcSwap`). First @@ -36,10 +24,9 @@ use arc_swap::ArcSwap; use rbitcoin_primitives::Fk; use rbitcoin_store::{OutputRecord, StoreError, TxRecord}; use std::cell::RefCell; -use std::collections::HashMap; use std::hash::BuildHasherDefault; use std::sync::atomic::{AtomicU8, Ordering}; -use std::sync::{Arc, Mutex, OnceLock, Weak}; +use std::sync::{Arc, OnceLock}; pub use rbitcoin_store::{FkMap, FkSet, U32Map, U64Map, U64Set}; @@ -53,7 +40,7 @@ const CB_TRUE: u8 = 2; /// Immutable pin outs (compose → publish, never mutate). #[derive(Debug, Clone)] enum PinOuts { - /// Plan / in-flight / RecentCreates: share the CreatePin Arc. + /// Plan / in-flight: share the CreatePin Arc. Full { pin: crate::CreatePin, checked: Vec, @@ -290,14 +277,13 @@ impl PinHalf { } } -/// One create's sparse pin payload, shared across concurrent pipeline batches. +/// One create's sparse pin payload, shared as `Arc` within a batch. /// /// Outs and layout are independent immutable Arc halves (compose only the half /// that changes). Vacant insert is Frozen; first real compose promotes to /// ArcSwap (lock-free load; RCU store). #[derive(Debug)] pub struct SharedParentPin { - fk: Fk, tx: TxRecord, /// 0 unknown, 1 not coinbase, 2 coinbase. coinbase: AtomicU8, @@ -309,7 +295,6 @@ pub struct SharedParentPin { impl SharedParentPin { fn new( - fk: Fk, tx: TxRecord, live: Vec<(u32, OutputRecord)>, checked: Vec, @@ -323,7 +308,6 @@ impl SharedParentPin { None => CB_UNKNOWN, }; Self { - fk, tx, coinbase: AtomicU8::new(cb), outs: PinHalf::new(PinOuts::new(live, checked)), @@ -332,7 +316,6 @@ impl SharedParentPin { } fn new_full( - fk: Fk, pin: crate::CreatePin, checked: Vec, coinbase: Option, @@ -346,7 +329,6 @@ impl SharedParentPin { }; let tx = pin.0.clone(); Self { - fk, tx, coinbase: AtomicU8::new(cb), outs: PinHalf::new(PinOuts::full(pin, checked)), @@ -474,185 +456,12 @@ impl SharedParentPin { } } -/// Prep-time registry: Weak map so dead pins free when last batch Arc drops. -/// -/// Mutex is only for bulk adopt / publish of Arc handles — never held while -/// assemble walks inputs or write fills layout data, and **not** on the -/// per-parent insert hot path. -#[derive(Debug, Default)] -struct PinIndex { - by_fk: U64Map>, - by_txid: HashMap<[u8; 32], Weak>, - /// Strong pins at last insert/gc (not a drop hook). Snapshot must not walk. - live: usize, -} - -/// Prep-time registry: Weak map so dead pins free when last batch Arc drops. -/// -/// Mutex is only for bulk adopt / publish / [`Self::bulk_lookup_txid`] — never -/// held while assemble walks inputs or write fills layout, and **not** on the -/// per-parent insert hot path. -#[derive(Debug, Default)] -pub struct PipelineParentStore { - maps: Mutex, -} - -impl PipelineParentStore { - pub fn new() -> Self { - Self { - maps: Mutex::new(PinIndex::default()), - } - } - - /// Live pin with non-zero txid and a stamped `txout` body range. - /// - /// Same Weak lifetime as outs share: last batch `Arc` drop → `None`. - /// Zero txid is never indexed. Live pin without `body_range` is a miss - /// (do not half-skip head). - pub fn lookup_txid(&self, txid: &[u8; 32]) -> Option<(Fk, (u64, u64))> { - self.bulk_lookup_txid(std::iter::once(txid)) - .into_iter() - .next() - .map(|(_, hit)| hit) - } - - /// One lock: [`Self::lookup_txid`] rules for every key. - /// - /// Dead Weaks are dropped from the txid index. Keys with no live pin, - /// zero txid, or no `body_range` are omitted from the map. - pub fn bulk_lookup_txid<'a>( - &self, - txids: impl IntoIterator, - ) -> HashMap<[u8; 32], (Fk, (u64, u64))> { - let mut g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - let mut out = HashMap::new(); - for txid in txids { - if *txid == [0u8; 32] { - continue; - } - let Some(w) = g.by_txid.get(txid) else { - continue; - }; - match w.upgrade() { - Some(p) => { - if let Some(r) = p.load_layout().body_range { - out.insert(*txid, (p.fk, r)); - } - } - None => { - g.by_txid.remove(txid); - } - } - } - out - } - - /// Live strong pins still reachable via Weak (diagnostics / tests). - pub fn live_count(&self) -> usize { - let g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - g.by_fk.values().filter(|w| w.strong_count() > 0).count() - } - - /// Occupancy: weak map slots, live strong pins, approx bytes of live pin outs. - pub fn size_snapshot(&self) -> (usize, usize, u64) { - let g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - let weak_slots = g.by_fk.len(); - let live = g.live.min(weak_slots); - let bytes = (weak_slots as u64) - .saturating_mul(24) - .saturating_add((live as u64).saturating_mul(256)); - (weak_slots, live, bytes) - } - - /// Drop dead Weaks now (keeps map from retaining empty slots after pin drop). - pub fn gc_dead_weaks(&self) { - let mut g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - g.by_fk.retain(|_, w| w.strong_count() > 0); - g.by_txid.retain(|_, w| w.strong_count() > 0); - g.by_fk.shrink_to_fit(); - g.by_txid.shrink_to_fit(); - g.live = g.by_fk.len(); - } - - /// One lock: upgrade live pins for `ids` into a map (prep batch start). - pub(crate) fn bulk_upgrade( - &self, - ids: impl IntoIterator, - ) -> U64Map> { - let g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - let mut out = U64Map::default(); - for id in ids { - if let Some(p) = g.by_fk.get(&id).and_then(|w| w.upgrade()) { - out.insert(id, p); - } - } - out - } - - /// One lock: publish **selected** batch pins as Weaks (new Arc registrations). - /// - /// `publish_ids` should list create_fks whose local Arc is new to the store - /// (typically vacant `insert_owned` results). Pure adopt hits already have a - /// Weak and need not be re-walked — full-map publish was O(all parents). - /// Does **not** retain-walk the Weak map (IBD load pin is the caller). Dead - /// slots are dropped by [`Self::gc_dead_weaks`]. - /// - /// On Arc identity conflict (peer batch won the slot), merge local sparse - /// fields into the existing Arc and replace the batch handle so both batches - /// share one payload. - pub(crate) fn bulk_publish_ids( - &self, - pins: &mut U64Map>, - publish_ids: &[u64], - ) { - if publish_ids.is_empty() { - return; - } - let mut conflicts: Vec<(u64, Arc, Arc)> = Vec::new(); - { - let mut g = self.maps.lock().unwrap_or_else(|e| e.into_inner()); - for &id in publish_ids { - let Some(pin) = pins.get(&id) else { - continue; - }; - match g.by_fk.get(&id).and_then(|w| w.upgrade()) { - Some(existing) if !Arc::ptr_eq(&existing, pin) => { - conflicts.push((id, existing, Arc::clone(pin))); - } - Some(_) => {} - None => { - let w = Arc::downgrade(pin); - g.by_fk.insert(id, w.clone()); - g.live = g.live.saturating_add(1); - if pin.tx.txid != [0u8; 32] { - g.by_txid.insert(pin.tx.txid, w); - } - } - } - } - } - for (id, existing, local) in conflicts { - let src_outs = local.load_outs(); - let src_lay = local.load_layout(); - existing.merge_outs(src_outs.sparse_live(), src_outs.checked()); - existing.set_coinbase_if_known(local.coinbase_opt()); - existing.maybe_merge_layout(src_lay.body_range, &src_lay.spender_rels); - pins.insert(id, existing); - } - } -} - /// Per-batch handle map: `create_fk → Arc` shared pin (refcount only on clone). /// /// Assemble sticky (`sticky_outs`) is batch-local and not shared across clones. #[derive(Debug, Default)] pub struct BatchParents { - /// Optional pipeline store for sharing across concurrent batches. - store: Option>, pins: U64Map>, - /// create_fks that need Weak registration (new Arc from this batch). - /// Pure adopt hits are omitted — already published by a prior batch. - publish_ids: Vec, /// Last outs Arc loaded for assemble (`get_parent_txout_parts`). sticky_outs: RefCell)>>, } @@ -660,10 +469,7 @@ pub struct BatchParents { impl Clone for BatchParents { fn clone(&self) -> Self { Self { - store: self.store.clone(), pins: self.pins.clone(), - // Cloned batch is a new handle map; re-publish if store-attached. - publish_ids: self.pins.keys().copied().collect(), sticky_outs: RefCell::new(None), } } @@ -672,32 +478,14 @@ impl Clone for BatchParents { impl BatchParents { pub fn new() -> Self { Self { - store: None, pins: U64Map::default(), - publish_ids: Vec::new(), sticky_outs: RefCell::new(None), } } pub fn with_capacity(n: usize) -> Self { Self { - store: None, pins: U64Map::with_capacity_and_hasher(n, BuildHasherDefault::default()), - publish_ids: Vec::with_capacity(n), - sticky_outs: RefCell::new(None), - } - } - - /// Prep/IBD: share pins with other batches via `store`. - /// - /// Inserts stay local; call [`adopt_from_store`] before pin fill and - /// [`publish_to_store`] after so the Weak registry stays current without a - /// per-parent mutex on the free-plan path. - pub fn with_store(store: Arc, capacity: usize) -> Self { - Self { - store: Some(store), - pins: U64Map::with_capacity_and_hasher(capacity, BuildHasherDefault::default()), - publish_ids: Vec::with_capacity(capacity), sticky_outs: RefCell::new(None), } } @@ -718,38 +506,7 @@ impl BatchParents { self.pins.is_empty() } - /// Bulk-adopt live shared pins for `ids` (one store lock). Call before pin fill. - pub fn adopt_from_store(&mut self, ids: impl IntoIterator) { - let Some(store) = &self.store else { - return; - }; - let upgraded = store.bulk_upgrade(ids); - if upgraded.is_empty() { - return; - } - self.pins.reserve(upgraded.len()); - for (id, pin) in upgraded { - self.pins.entry(id).or_insert(pin); - } - } - - /// Bulk-publish **new** local pins into the pipeline store (one store lock). - /// Call after pin fill. Adopted Arcs are already registered and skipped. - pub fn publish_to_store(&mut self) { - let Some(store) = &self.store else { - self.publish_ids.clear(); - return; - }; - if self.publish_ids.is_empty() { - return; - } - self.publish_ids.sort_unstable(); - self.publish_ids.dedup(); - store.bulk_publish_ids(&mut self.pins, &self.publish_ids); - self.publish_ids.clear(); - } - - /// Layout / coinbase only when outs already cover need (cross-batch share hit). + /// Layout / coinbase only when outs already cover need (share hit). /// /// Skips all work when there is no meta material (pure share hit). #[inline] @@ -774,10 +531,8 @@ impl BatchParents { /// Insert / merge one parent (prep pin hot path). /// - /// **No store mutex** — pure batch HashMap. First insert for an id is the - /// pre-share cost class (`ParentEntry` put). Merge only if the same batch - /// already holds a partial pin (or after adopt left an incomplete cover). - /// Occupied path uses one snap decision for outs+layout (single-snap). + /// Pure batch HashMap. Merge only if the same batch already holds a partial + /// pin. Occupied path uses one snap decision for outs+layout (single-snap). #[inline] pub fn insert_owned( &mut self, @@ -814,7 +569,6 @@ impl BatchParents { } std::collections::hash_map::Entry::Vacant(v) => { v.insert(Arc::new(SharedParentPin::new( - fk, tx, live, checked, @@ -822,8 +576,6 @@ impl BatchParents { body_range, spender_rels, ))); - // New Arc — must register Weak on publish (adopt hits skip this). - self.publish_ids.push(id); } } } @@ -868,14 +620,12 @@ impl BatchParents { } std::collections::hash_map::Entry::Vacant(v) => { v.insert(Arc::new(SharedParentPin::new_full( - fk, pin, checked, coinbase, body_range, spender_rels, ))); - self.publish_ids.push(id); } } } @@ -1214,10 +964,6 @@ impl BatchParents { return; } self.pins.reserve(other.pins.len()); - // Carry forward other's unpublished Weak-registration set so a later - // publish_to_store still registers Arcs that only lived on `other`. - self.publish_ids.reserve(other.publish_ids.len()); - self.publish_ids.extend(other.publish_ids.iter().copied()); for (id, src) in other.pins { match self.pins.entry(id) { std::collections::hash_map::Entry::Vacant(v) => { @@ -1559,168 +1305,6 @@ mod tests { assert!(bp.get_spender_abs(Fk(1), 1).is_none()); } - /// Two batches with the same store share one SharedParentPin after publish/adopt. - #[test] - fn pipeline_store_shares_one_arc_across_batches() { - let store = Arc::new(PipelineParentStore::new()); - let mut a = BatchParents::with_store(Arc::clone(&store), 4); - let mut b = BatchParents::with_store(Arc::clone(&store), 4); - a.insert_owned( - Fk(7), - tx(7), - vec![(0, out(10))], - vec![0], - Some(false), - None, - vec![(0, 5)], - ); - a.publish_to_store(); - b.adopt_from_store([7]); - b.insert_owned( - Fk(7), - tx(7), - vec![(1, out(20))], - vec![1], - None, - None, - Vec::new(), - ); - b.publish_to_store(); - let pa = a.pins.get(&7).expect("a has pin"); - let pb = b.pins.get(&7).expect("b has pin"); - assert!( - Arc::ptr_eq(pa, pb), - "batches must share one SharedParentPin Arc after adopt" - ); - assert!(a.has_parent_out(Fk(7), 0)); - assert!(a.has_parent_out(Fk(7), 1), "merged vout 1 visible via a"); - assert!(b.has_parent_out(Fk(7), 0), "merged vout 0 visible via b"); - assert!(b.has_parent_out(Fk(7), 1)); - assert_eq!(store.live_count(), 1); - drop(a); - assert_eq!(store.live_count(), 1, "b still holds pin"); - drop(b); - assert_eq!(store.live_count(), 0, "last batch drop releases pin"); - } - - /// Concurrent local inserts then publish: loser merges into winner Arc. - #[test] - fn bulk_publish_merges_race_to_one_arc() { - let store = Arc::new(PipelineParentStore::new()); - let mut a = BatchParents::with_store(Arc::clone(&store), 2); - let mut b = BatchParents::with_store(Arc::clone(&store), 2); - a.insert_owned( - Fk(1), - tx(1), - vec![(0, out(1))], - vec![0], - None, - None, - vec![(0, 1)], - ); - b.insert_owned( - Fk(1), - tx(1), - vec![(1, out(2))], - vec![1], - None, - None, - vec![(1, 2)], - ); - // a publishes first (wins Weak slot). - a.publish_to_store(); - // b publishes: must merge into a's Arc and swap handle. - b.publish_to_store(); - let pa = a.pins.get(&1).unwrap(); - let pb = b.pins.get(&1).unwrap(); - assert!(Arc::ptr_eq(pa, pb)); - assert!(a.has_parent_out(Fk(1), 0)); - assert!(a.has_parent_out(Fk(1), 1)); - assert!(b.has_parent_out(Fk(1), 0)); - assert!(b.has_parent_out(Fk(1), 1)); - assert_eq!(store.live_count(), 1); - } - - /// extend_from must carry publish_ids so vacant pins from `other` still register. - #[test] - fn extend_from_merges_publish_ids_for_store_registration() { - let store = Arc::new(PipelineParentStore::new()); - let mut a = BatchParents::with_store(Arc::clone(&store), 8); - let mut b = BatchParents::with_store(Arc::clone(&store), 8); - a.insert_owned( - Fk(1), - tx(1), - vec![(0, out(1))], - vec![0], - Some(false), - Some((8, 16)), - vec![(0, 4)], - ); - b.insert_owned( - Fk(2), - tx(2), - vec![(0, out(2))], - vec![0], - Some(false), - Some((24, 16)), - vec![(0, 4)], - ); - a.extend_from(b); - assert_eq!(a.len(), 2); - a.publish_to_store(); - assert_eq!( - store.live_count(), - 2, - "both fks must register: extend_from must merge publish_ids" - ); - assert!(a.contains(Fk(1))); - assert!(a.contains(Fk(2))); - } - - /// Adopted pins are already Weak-registered; publish only registers vacant inserts. - #[test] - fn publish_registers_only_new_not_pure_adopt() { - let store = Arc::new(PipelineParentStore::new()); - let mut seed = BatchParents::with_store(Arc::clone(&store), 16); - for i in 1..=10u64 { - seed.insert_owned( - Fk(i), - tx(i as u8), - vec![(0, out(i as i64))], - vec![0], - Some(false), - Some((i * 8, 16)), - vec![(0, 4)], - ); - } - seed.publish_to_store(); - assert_eq!(store.live_count(), 10); - - let mut bp = BatchParents::with_store(Arc::clone(&store), 16); - bp.adopt_from_store(1..=10); - assert_eq!(bp.len(), 10); - // Pure adopt: publish is a no-op (no new Arcs). - bp.publish_to_store(); - assert_eq!(store.live_count(), 10); - - // Vacant insert of a new fk registers on publish. - bp.insert_owned( - Fk(99), - tx(99), - vec![(0, out(99))], - vec![0], - Some(false), - Some((99 * 8, 16)), - vec![(0, 4)], - ); - bp.publish_to_store(); - assert_eq!(store.live_count(), 11); - assert!(bp.contains(Fk(99))); - // Second publish with no new inserts is free. - bp.publish_to_store(); - assert_eq!(store.live_count(), 11); - } - /// Identity hasher for pack-scale u64 keys is the raw key (no SipHash mix). /// Write/lookup structural maps depend on this for the measured CPU win. #[test] @@ -1746,28 +1330,6 @@ mod tests { assert_eq!(m.get(&(n + 1)), None); } - /// Free-plan insert must not require a store hit — vacant path is local only. - #[test] - fn insert_owned_local_without_publish_leaves_store_empty() { - let store = Arc::new(PipelineParentStore::new()); - let mut bp = BatchParents::with_store(Arc::clone(&store), 8); - for i in 1..=100u64 { - bp.insert_owned( - Fk(i), - tx((i % 200) as u8), - vec![(0, out(i as i64))], - vec![0], - Some(false), - None, - Vec::new(), - ); - } - assert_eq!(bp.len(), 100); - assert_eq!(store.live_count(), 0, "insert must not touch store"); - bp.publish_to_store(); - assert_eq!(store.live_count(), 100); - } - #[test] fn vacant_insert_does_not_arcswap_until_compose() { let mut bp = BatchParents::new(); @@ -1902,8 +1464,7 @@ mod tests { use std::sync::Barrier; use std::thread; - let store = Arc::new(PipelineParentStore::new()); - let mut writer = BatchParents::with_store(Arc::clone(&store), 1); + let mut writer = BatchParents::with_capacity(1); writer.insert_owned( Fk(1), tx(1), @@ -1913,7 +1474,6 @@ mod tests { None, vec![(0, 10)], ); - writer.publish_to_store(); let pin = Arc::clone(writer.pins.get(&1).expect("shared pin")); let barrier = Arc::new(Barrier::new(2)); @@ -1960,9 +1520,8 @@ mod tests { #[test] fn pin_compose_multi_pack_timed() { let n_parents = 8_000usize; // ~input budget scale - let store = Arc::new(PipelineParentStore::new()); let t0 = std::time::Instant::now(); - let mut a = BatchParents::with_store(Arc::clone(&store), n_parents); + let mut a = BatchParents::with_capacity(n_parents); for i in 1..=n_parents as u64 { a.insert_owned( Fk(i), @@ -1974,15 +1533,12 @@ mod tests { vec![(0, 10)], ); } - a.publish_to_store(); let insert_ns = t0.elapsed().as_nanos(); - // Covered re-insert (production free-pin share hit after adopt). + // Covered re-insert (Occupied no-op outs). let t_cov = std::time::Instant::now(); - let mut cov = BatchParents::with_store(Arc::clone(&store), n_parents); - cov.adopt_from_store(1..=n_parents as u64); for i in 1..=n_parents as u64 { - cov.insert_owned( + a.insert_owned( Fk(i), tx((i % 200) as u8), vec![(0, out(i as i64))], @@ -1997,23 +1553,21 @@ mod tests { // Layout-only fill (write ensure path) — same denserels shape as baseline. let t_lay = std::time::Instant::now(); for i in 1..=n_parents as u64 { - cov.set_layout_for_need(Fk(i), (i * 100, 50), &[10], &[]); + a.set_layout_for_need(Fk(i), (i * 100, 50), &[10], &[]); } let layout_ns = t_lay.elapsed().as_nanos(); // Second ensure pass — same API; already_covers short-circuit. let t_lay2 = std::time::Instant::now(); for i in 1..=n_parents as u64 { - cov.set_layout_for_need(Fk(i), (i * 100, 50), &[10], &[]); + a.set_layout_for_need(Fk(i), (i * 100, 50), &[10], &[]); } let layout2_ns = t_lay2.elapsed().as_nanos(); let t1 = std::time::Instant::now(); - let mut b = BatchParents::with_store(Arc::clone(&store), n_parents); - b.adopt_from_store(1..=n_parents as u64); for i in 1..=n_parents as u64 { - // Widen need + layout (compose publish on shared pins). - b.insert_owned( + // Widen need + layout (compose publish on the same pins). + a.insert_owned( Fk(i), tx((i % 200) as u8), vec![(1, out(i as i64 + 1))], @@ -2023,10 +1577,9 @@ mod tests { vec![(1, 20)], ); } - b.publish_to_store(); let widen_ns = t1.elapsed().as_nanos(); - // Multi-input same-parent: vouts 0 and 1 after widen (shared Arc on `a`). + // Multi-input same-parent: vouts 0 and 1 after widen. // Fair cold = same `parent_txout_parts` path with sticky disabled. let reps = 10usize; let n_inputs = n_parents * reps * 2; @@ -2058,7 +1611,6 @@ mod tests { assert_eq!(sum, sum_c); assert_eq!(a.len(), n_parents); - assert_eq!(b.len(), n_parents); assert!(a.pin_covered(Fk(1), &[0, 1])); a.set_spent_range_only(Fk(1), (1000, 24)); assert_eq!(a.get_spender_abs(Fk(1), 1), Some(1008)); @@ -2215,7 +1767,6 @@ mod tests { #[test] fn pin_body_compose_does_not_mutate_source() { let pin = SharedParentPin::new( - Fk(1), tx(1), vec![(0, out(10))], vec![0], @@ -2312,15 +1863,10 @@ mod tests { assert_eq!(bp.get_spender_abs(Fk(1), 1), Some(808)); } - /// PipelineParentStore size_snapshot / gc + set_layout_sparse / body_range_only edges. + /// `set_layout_sparse` / `set_body_range_only` edges on missing pins. #[test] - fn pipeline_store_snapshot_gc_and_layout_sparse_surface() { - let store = Arc::new(PipelineParentStore::new()); - assert_eq!(store.size_snapshot(), (0, 0, 0)); - assert_eq!(store.live_count(), 0); - store.gc_dead_weaks(); // empty map no-op - - let mut bp = BatchParents::with_store(Arc::clone(&store), 8); + fn layout_sparse_and_body_range_only_surface() { + let mut bp = BatchParents::with_capacity(8); // Null / missing pin early-outs (set_layout_sparse + set_body_range_only). bp.set_layout_sparse(Fk::NULL, (0, 10), vec![(0, 1)], &[]); bp.set_layout_sparse(Fk(99), (0, 10), vec![(0, 1)], &[0]); @@ -2347,17 +1893,6 @@ mod tests { Some((100, 40)), vec![(0, 5)], ); - bp.publish_to_store(); - let (weak, live, bytes) = store.size_snapshot(); - assert_eq!(weak, 2); - assert_eq!(live, 2); - assert_eq!( - bytes, - (weak as u64) - .saturating_mul(24) - .saturating_add((live as u64).saturating_mul(256)), - "O(1) slot estimate — must not walk pin scripts" - ); // Grow checked via extra_need, then fill sparse layout. bp.set_layout_sparse(Fk(1), (200, 50), vec![(0, 7), (1, 8)], &[1]); @@ -2375,18 +1910,6 @@ mod tests { assert_eq!(bp.get_body_range(Fk(2)), Some((300, 60))); bp.set_body_range_only(Fk(2), (300, 60)); assert_eq!(bp.get_body_range(Fk(2)), Some((300, 60))); - - // Drop all strong refs → Weaks die; gc shrinks map. - drop(bp); - assert_eq!(store.live_count(), 0); - let (weak_dead, live_snap, _) = store.size_snapshot(); - assert!(weak_dead >= 2, "dead Weaks still occupy slots until gc"); - let _ = live_snap; // O(1) snapshot may lag until gc (no slot walk) - store.gc_dead_weaks(); - let (weak_after, live_after, bytes_after) = store.size_snapshot(); - assert_eq!(weak_after, 0); - assert_eq!(live_after, 0); - assert_eq!(bytes_after, 0); } #[test] @@ -2418,173 +1941,4 @@ mod tests { assert!(bp.get_parent_coinbase(Fk::NULL).is_none()); assert!(bp.get_body_range(Fk::NULL).is_none()); } - - #[test] - fn pipeline_parent_store_lookup_txid() { - let store = Arc::new(PipelineParentStore::new()); - let mut bp = BatchParents::with_store(Arc::clone(&store), 1); - let tid = tx(7).txid; - bp.insert_owned( - Fk(42), - tx(7), - vec![(0, out(1))], - vec![0], - Some(false), - Some((1000, 80)), - Vec::new(), - ); - bp.publish_to_store(); - assert_eq!( - store.lookup_txid(&tid), - Some((Fk(42), (1000, 80))), - "live pin with range must resolve by txid" - ); - let mut zero = tx(8); - zero.txid = [0u8; 32]; - let mut bp0 = BatchParents::with_store(Arc::clone(&store), 1); - bp0.insert_owned( - Fk(43), - zero, - vec![(0, out(1))], - vec![0], - Some(false), - Some((2000, 10)), - Vec::new(), - ); - bp0.publish_to_store(); - assert!( - store.lookup_txid(&[0u8; 32]).is_none(), - "zero txid is never indexed" - ); - drop(bp); - drop(bp0); - store.gc_dead_weaks(); - assert!( - store.lookup_txid(&tid).is_none(), - "txid index must die with the last pin Arc" - ); - } - - /// One lock: live + range hits; dead Weak, missing range, zero txid, and - /// unknown keys miss — same as N× [`PipelineParentStore::lookup_txid`]. - #[test] - fn pipeline_parent_store_bulk_lookup_txid() { - let store = Arc::new(PipelineParentStore::new()); - let live_tid = tx(1).txid; - let dead_tid = tx(2).txid; - let no_range_tid = tx(3).txid; - let missing_tid = tx(4).txid; - let zero = [0u8; 32]; - - let mut live = BatchParents::with_store(Arc::clone(&store), 2); - live.insert_owned( - Fk(10), - tx(1), - vec![(0, out(1))], - vec![0], - Some(false), - Some((1000, 80)), - Vec::new(), - ); - let mut no_range = BatchParents::with_store(Arc::clone(&store), 1); - no_range.insert_owned( - Fk(11), - tx(3), - vec![(0, out(1))], - vec![0], - Some(false), - None, - Vec::new(), - ); - let mut dead = BatchParents::with_store(Arc::clone(&store), 1); - dead.insert_owned( - Fk(12), - tx(2), - vec![(0, out(1))], - vec![0], - Some(false), - Some((2000, 40)), - Vec::new(), - ); - live.publish_to_store(); - no_range.publish_to_store(); - dead.publish_to_store(); - drop(dead); - - let keys = [live_tid, dead_tid, no_range_tid, missing_tid, zero]; - let expected: Vec> = - keys.iter().map(|t| store.lookup_txid(t)).collect(); - let bulk = store.bulk_lookup_txid(keys.iter()); - for (t, exp) in keys.iter().zip(expected.iter()) { - assert_eq!( - bulk.get(t).copied(), - *exp, - "bulk must match per-key lookup for {t:?}" - ); - } - assert_eq!(bulk.get(&live_tid).copied(), Some((Fk(10), (1000, 80)))); - assert!(bulk.get(&dead_tid).is_none()); - assert!(bulk.get(&no_range_tid).is_none()); - assert!(bulk.get(&missing_tid).is_none()); - assert!(bulk.get(&zero).is_none()); - let _keep = (live, no_range); - } - - fn tx_n(n: u64) -> TxRecord { - let mut txid = [0u8; 32]; - txid[..8].copy_from_slice(&n.to_le_bytes()); - TxRecord { - txid, - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 1, - output_start_fk: Fk::NULL, - output_count: 1, - } - } - - /// Publish must not walk/retain the Weak map. Dead slots stay until - /// [`PipelineParentStore::gc_dead_weaks`] (mem-stats), not every insert. - #[test] - fn bulk_publish_does_not_retain_dead_weaks() { - let store = Arc::new(PipelineParentStore::new()); - let n = 16_385usize; - let mut bp = BatchParents::with_store(Arc::clone(&store), n); - for i in 1..=n as u64 { - bp.insert_owned( - Fk(i), - tx_n(i), - vec![(0, out(1))], - vec![0], - Some(false), - None, - Vec::new(), - ); - } - bp.publish_to_store(); - drop(bp); - assert_eq!(store.live_count(), 0); - - let mut next = BatchParents::with_store(Arc::clone(&store), 1); - next.insert_owned( - Fk(n as u64 + 1), - tx_n(n as u64 + 1), - vec![(0, out(2))], - vec![0], - Some(false), - None, - Vec::new(), - ); - next.publish_to_store(); - let (weak, _, _) = store.size_snapshot(); - assert!( - weak >= n, - "publish must not retain-walk dead Weaks (weak={weak}, need>={n})" - ); - assert_eq!(store.live_count(), 1); - drop(next); - store.gc_dead_weaks(); - assert_eq!(store.size_snapshot().0, 0); - } } diff --git a/crates/rbitcoin-query/src/in_flight.rs b/crates/rbitcoin-query/src/in_flight.rs index 64ec9a90..5f670ca8 100644 --- a/crates/rbitcoin-query/src/in_flight.rs +++ b/crates/rbitcoin-query/src/in_flight.rs @@ -6,17 +6,17 @@ //! never mutated — no `Arc::make_mut` of a shared whole-map while prep holds a //! snapshot. //! -//! **Prune:** drop a layer only when **both** drain has inserted its create-fk -//! span (`fk_hi <= drain_fk`) **and** the fence covers that span. TipOnly is -//! fence-connected; drain alone is not a home. Either signal alone keeps the -//! layer (Class C ∥ drain/seal). Call after pin — stamp skips `body_range` -//! when this map still has CreatePin outs. +//! **Prune:** drain inserted the create-fk span **and** the fence covers it. +//! Pin layers (`outs` non-empty, tagged `max_height`) also wait for write +//! [`InFlightLog::set_keep_until`] (`until = lookup_started_hi.max(hi)`) and +//! `class_a_hi >= until`. Creates-only / +//! untagged layers still drop at drain+fence (TipOnly home). Drain or fence +//! alone keeps the layer. Call after pin so n−1 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. +//! Lookup is newest→oldest scan over layers (O(L)). use crate::archive::CreatePin; -use crate::U64Map; +use crate::{U32Map, U64Map}; use rbitcoin_primitives::Fk; use std::collections::HashMap; use std::sync::Arc; @@ -124,11 +124,16 @@ fn layer_approx_bytes(creates: &HashMap<[u8; 32], Fk>, outs: &U64Map) #[derive(Debug, Default, Clone)] pub struct InFlightLog { layers: Vec>, + /// Write-frozen keep horizon: pack `max_height` → `until`. + keep_until: U32Map, } impl InFlightLog { pub fn new() -> Self { - Self { layers: Vec::new() } + Self { + layers: Vec::new(), + keep_until: U32Map::default(), + } } /// Publish one pack. Does not mutate existing layers. @@ -151,20 +156,40 @@ impl InFlightLog { Some(h) => h < height, None => true, }); + self.keep_until.retain(|&h, _| h < height); self.layers.shrink_to_fit(); } + /// Write-time keep horizon (`until = lookup_started_hi.max(height)`). + pub fn set_keep_until(&mut self, height: u32, until: u32) { + self.keep_until + .entry(height) + .and_modify(|u| *u = (*u).max(until)) + .or_insert(until); + } + + pub fn apply_keep_untils(&mut self, rows: impl IntoIterator) { + for (h, u) in rows { + self.set_keep_until(h, u); + } + } + /// Drop layers TipOnly can see: inserted **and** fenced. /// - /// Call **after** pin (scripts handoff) so n−1 still has CreatePin outs. - /// Stamp skips `body_range` when `get_out` hits. `drain_fk == 0` keeps - /// every layer. Empty-span layers stay. - pub fn prune_if_head_ready(&mut self, fence: &rbitcoin_store::HeightFence, drain_fk: u64) { + /// Pin layers with a keep-until also need `class_a_hi >= until`. Untagged + /// / creates-only drop at drain+fence. `drain_fk == 0` keeps every layer. + pub fn prune_if_head_ready( + &mut self, + fence: &rbitcoin_store::HeightFence, + drain_fk: u64, + class_a_hi: Option, + ) { if self.layers.is_empty() || drain_fk == 0 { return; } + let keep_until = &self.keep_until; self.layers - .retain(|layer| !layer.head_ready(fence, drain_fk)); + .retain(|layer| !layer_drop_ready(layer, fence, drain_fk, class_a_hi, keep_until)); self.layers.shrink_to_fit(); } @@ -188,6 +213,7 @@ impl InFlightLog { pub fn clear(&mut self) { self.layers.clear(); + self.keep_until.clear(); self.layers.shrink_to_fit(); } @@ -217,6 +243,25 @@ impl InFlightLog { } } +fn layer_drop_ready( + layer: &InFlightLayer, + fence: &rbitcoin_store::HeightFence, + drain_fk: u64, + class_a_hi: Option, + keep_until: &U32Map, +) -> bool { + if !layer.head_ready(fence, drain_fk) { + return false; + } + let Some(h) = layer.max_height else { + return true; + }; + if let Some(&until) = keep_until.get(&h) { + return class_a_hi.is_some_and(|c| c >= until); + } + layer.outs.is_empty() +} + /// Immutable prep/plan view over published layers (newest→oldest lookup). #[derive(Debug, Clone)] pub struct InFlightView { @@ -410,25 +455,58 @@ mod tests { let covered = fence_covering(10, 1, 2); let empty = rbitcoin_store::HeightFence::empty(); - log.prune_if_head_ready(&covered, 0); + log.prune_if_head_ready(&covered, 0, Some(2)); assert!( log.snapshot().get_create_fk(&p.0.txid).is_some(), "drain_fk 0: insert never completed" ); - log.prune_if_head_ready(&covered, 9); + log.prune_if_head_ready(&covered, 9, Some(2)); assert!( log.snapshot().get_create_fk(&p.0.txid).is_some(), "fence covers, drain still mid-seal" ); - log.prune_if_head_ready(&empty, 10); + log.prune_if_head_ready(&empty, 10, Some(2)); assert!( log.snapshot().get_create_fk(&p.0.txid).is_some(), "drain done, fence not connected — TipOnly would drop" ); - log.prune_if_head_ready(&covered, 10); + log.set_keep_until(2, 2); + log.prune_if_head_ready(&covered, 10, Some(2)); + assert!( + log.snapshot().get_create_fk(&p.0.txid).is_none(), + "inserted, fenced, and class_a covers until" + ); + } + + #[test] + fn prune_keeps_pin_layer_until_class_a_covers_keep_until() { + let mut log = InFlightLog::new(); + let p = pin(10); + log.note_layer(InFlightLayer::from_plan_pins([(Fk(10), &p)]).with_max_height(2)); + let covered = fence_covering(10, 1, 2); + log.set_keep_until(2, 40); + log.prune_if_head_ready(&covered, 10, Some(39)); + assert!( + log.snapshot().get_create_fk(&p.0.txid).is_some(), + "drain+fence is not enough while class_a_hi < until" + ); + log.prune_if_head_ready(&covered, 10, Some(40)); assert!( log.snapshot().get_create_fk(&p.0.txid).is_none(), - "inserted and fenced" + "class_a_hi >= until drops after drain+fence" + ); + } + + #[test] + fn prune_keeps_tagged_pin_layer_until_write_stamps_keep_until() { + let mut log = InFlightLog::new(); + let p = pin(10); + log.note_layer(InFlightLayer::from_plan_pins([(Fk(10), &p)]).with_max_height(2)); + let covered = fence_covering(10, 1, 2); + log.prune_if_head_ready(&covered, 10, Some(2)); + assert!( + log.snapshot().get_create_fk(&p.0.txid).is_some(), + "max_height pin layer waits for write keep-until" ); } @@ -455,7 +533,7 @@ mod tests { let b = pin(50); log.note_layer(InFlightLayer::from_plan_pins([(Fk(10), &a)])); log.note_layer(InFlightLayer::from_plan_pins([(Fk(50), &b)])); - log.prune_if_head_ready(&fence_covering(1, 50, 2), 10); + log.prune_if_head_ready(&fence_covering(1, 50, 2), 10, None); let v = log.snapshot(); assert!(v.get_create_fk(&a.0.txid).is_none()); assert!( diff --git a/crates/rbitcoin-query/src/layer_chain.rs b/crates/rbitcoin-query/src/layer_chain.rs index d2ac89a3..d84d03c5 100644 --- a/crates/rbitcoin-query/src/layer_chain.rs +++ b/crates/rbitcoin-query/src/layer_chain.rs @@ -1,6 +1,5 @@ -//! Newest-first Arc layer list. live_union and RecentCreates share splice/prepend. +//! Newest-first Arc layer list. live_union uses splice/prepend. -use arc_swap::{ArcSwapOption, Guard}; use std::sync::Arc; /// One immutable layer plus the older chain. `hits` is never cloned on splice. @@ -82,68 +81,3 @@ pub fn splice_kept( } new_head } - -fn option_arc_eq(a: &Option>, b: &Option>) -> bool { - match (a, b) { - (None, None) => true, - (Some(x), Some(y)) => Arc::ptr_eq(x, y), - _ => false, - } -} - -/// Store `next` when `slot` still holds `expected`. Lost CAS returns the live head. -pub fn cas_head( - slot: &ArcSwapOption>, - expected: &Option>>, - next: Option>>, -) -> Result<(), Option>>> { - let prev = Guard::into_inner(slot.compare_and_swap(expected, next)); - if option_arc_eq(expected, &prev) { - Ok(()) - } else { - Err(prev) - } -} - -/// Apply `f` to the live head until [`cas_head`] lands. -pub fn rcu_head( - slot: &ArcSwapOption>, - mut f: impl FnMut(&Option>>) -> Option>>, -) { - let mut cur = slot.load_full(); - loop { - let next = f(&cur); - match cas_head(slot, &cur, next) { - Ok(()) => return, - Err(live) => cur = live, - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use arc_swap::ArcSwapOption; - - #[test] - fn cas_head_rejects_stale_expected() { - let slot: ArcSwapOption> = ArcSwapOption::empty(); - let a = ChainLayer::prepend(None, 10, 10, 10, Arc::new(1u8)); - assert!(cas_head(&slot, &None, Some(Arc::clone(&a))).is_ok()); - let stale = Some(Arc::clone(&a)); - rcu_head(&slot, |cur| splice_kept(cur.clone(), |l| l.lo < 10)); - assert!(slot.load_full().is_none()); - let c = ChainLayer::prepend(stale.clone(), 12, 12, 12, Arc::new(2u8)); - assert!( - cas_head(&slot, &stale, Some(c)).is_err(), - "CAS against a dropped head must fail" - ); - assert!(slot.load_full().is_none(), "stale prepend must not restore"); - rcu_head(&slot, |cur| { - Some(ChainLayer::prepend(cur.clone(), 12, 12, 12, Arc::new(2u8))) - }); - let head = slot.load_full().expect("retry"); - assert_eq!(*head.hits, 2); - assert!(head.older.is_none()); - } -} diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 782358fc..7feba346 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -11,7 +11,6 @@ mod connect; mod in_flight; mod layer_chain; mod published_ids; -mod recent_creates; mod reconstruct; mod resolved_wire; mod run_builder_core; @@ -69,7 +68,7 @@ pub struct ProcessOwnedSizes { pub inflight_layers: usize, pub inflight_pins: usize, pub inflight_bytes: u64, - /// PipelineParentStore Weak map + live strong pins. + /// Unused process pstore meters (always 0; BatchParents is batch-local). pub pstore_weak: usize, pub pstore_live: usize, pub pstore_bytes: u64, @@ -149,8 +148,8 @@ pub mod process_mem_stats { pub use archive::{ArchiveWritePlan, CreatePin}; pub use batch_parents::{ - layout_covers_need, sparse_spender_rels, BatchParents, FkMap, FkSet, PipelineParentStore, - SharedParentPin, U32Map, U64Map, U64Set, SPENDER_REL_UNKNOWN, + layout_covers_need, sparse_spender_rels, BatchParents, FkMap, FkSet, SharedParentPin, U32Map, + U64Map, U64Set, SPENDER_REL_UNKNOWN, }; pub use catchup::IndexMode; pub use chain_view::{ChainView, ChainViewKind}; @@ -161,7 +160,6 @@ pub use in_flight::{InFlightLayer, InFlightLog, InFlightView}; pub use published_ids::{ IdLayer, IdMap, LiveUnion, OutPointHasher, OutPointSet, PublishedIds, TxidHasher, }; -pub use recent_creates::RecentCreates; pub use scripthash::{ apply_history_filter, HistoryFilter, HistoryOrder, ScanUtxo, ScriptHashBalance, ScriptHashChainStats, ScriptHashHistoryItem, ScriptHashOutpoint, ScriptHashUtxo, ShJoinSlot, @@ -198,7 +196,7 @@ pub mod confirm_load_stats { pub static PIN_ADOPT_NS: AtomicU64 = AtomicU64::new(0); /// Post cold-range denserels: insert_owned into BatchParents (not IO). pub static PIN_RANGE_FILL_NS: AtomicU64 = AtomicU64::new(0); - /// RecentCreates create_pin probe (after in-flight / same-batch, before range fill). + /// Stamp-carried CreatePin probe (after in-flight / same-batch, before range fill). pub static PIN_RECENT_OUTS_NS: AtomicU64 = AtomicU64::new(0); /// Final pin contract (contains + pin_covered) wall. pub static PIN_CONTRACT_NS: AtomicU64 = AtomicU64::new(0); @@ -457,7 +455,7 @@ pub mod archive_phase_stats { pub static EXT_NEED: AtomicU64 = AtomicU64::new(0); pub static HEAD_NEED: AtomicU64 = AtomicU64::new(0); pub static HEAD_HIT: AtomicU64 = AtomicU64::new(0); - /// Unique prev_txids resolved from live [`crate::PipelineParentStore`]. + /// Unique prev_txids resolved from published live_union. pub static PIN_TXID_N: AtomicU64 = AtomicU64::new(0); /// Wall of that consult (RAM). pub static PIN_TXID_NS: AtomicU64 = AtomicU64::new(0); @@ -1169,9 +1167,8 @@ pub struct Query { disconnect_gen: AtomicU64, /// Lookup-published parent identity union (wave hits still in the BQ window). published_ids: std::sync::Arc, - /// Write-published just-confirmed identity (txid → fk+range). Load stamp - /// before leftover TipOnly. Outs are not stored. - recent_creates: std::sync::Arc, + /// Write-frozen in-flight keep-until (`height → until`), consumed at prune. + create_keep_until: Mutex>, } /// In-process hash→height map for the confirmed tip chain (~33 MiB raw at 1e6 tips). @@ -1261,7 +1258,7 @@ impl Query { disconnect_height: AtomicU32::new(0), disconnect_gen: AtomicU64::new(0), published_ids: std::sync::Arc::new(crate::PublishedIds::new()), - recent_creates: std::sync::Arc::new(crate::RecentCreates::new()), + create_keep_until: Mutex::new(crate::U32Map::default()), }; if let Some(tip) = q.tip_height() { let _ = q.ensure_height_by_hash_index(tip); @@ -1288,8 +1285,13 @@ impl Query { self.disconnect_height .store(height, AtomicOrdering::Release); self.disconnect_gen.fetch_add(1, AtomicOrdering::Release); - self.recent_creates.drop_from(height); - self.recent_creates.publish_if_dirty(); + { + let mut g = self + .create_keep_until + .lock() + .unwrap_or_else(|p| p.into_inner()); + g.retain(|&h, _| h < height); + } let rewind = if height == 0 { None } else { @@ -1654,95 +1656,25 @@ impl Query { &self.published_ids } - /// Just-confirmed identity ring (write-published; load stamp before leftover). - pub fn recent_creates(&self) -> &std::sync::Arc { - &self.recent_creates - } - - /// After Class A + idx: note identity rows and flush (layer until = - /// `lookup_started_hi`, drop when `class_a_hi` covers it). Missing idx - /// range is skipped (leftover TipOnly stays the home). - pub fn publish_recent_creates( - &self, - height: u32, - creates: impl IntoIterator, - ) -> Result<(), QueryError> { - self.note_recent_creates(height, creates)?; - self.expire_recent_creates(height); - Ok(()) - } - - /// Note identity rows and flush so a single-height caller sees `get`. - /// - /// Write batches use [`Self::note_recent_creates_defer`] + one - /// [`Self::flush_recent_creates`] so the live map is cloned once. - pub fn note_recent_creates( - &self, - height: u32, - creates: impl IntoIterator, - ) -> Result<(), QueryError> { - self.note_recent_creates_defer(height, creates)?; - self.flush_recent_creates(); - Ok(()) - } - - /// Note identity rows without rebuilding the load snapshot. - pub fn note_recent_creates_defer( - &self, - height: u32, - creates: impl IntoIterator, - ) -> Result<(), QueryError> { - let pairs: Vec<([u8; 32], Fk)> = creates.into_iter().collect(); - if pairs.is_empty() { - return Ok(()); - } - let fks: Vec = pairs.iter().map(|(_, fk)| *fk).collect(); - let ranges = self.store.tx_body_range_batch(&fks)?; - let rows = pairs - .into_iter() - .zip(ranges) - .filter_map(|((txid, fk), range)| range.map(|r| (txid, fk, r))); - self.recent_creates.note(height, rows); - Ok(()) - } - - /// Note already-ranged identity rows (write batches idx once, then note). - pub fn note_recent_creates_rows( - &self, - height: u32, - rows: impl IntoIterator, - ) { - self.recent_creates.note(height, rows); - } - - /// Same as [`Self::note_recent_creates_rows`] with optional [`CreatePin`] Arcs. - pub fn note_recent_creates_pins( - &self, - height: u32, - rows: impl IntoIterator)>, - ) { - self.recent_creates.note_pins(height, rows); - } - - /// Prepend pending notes as one layer (`until = started.max(hi)`) then drop - /// layers with `until <= class_a_hi`. - pub fn flush_recent_creates(&self) { - let started = self.lookup_started_hi().unwrap_or(0); - self.recent_creates.publish_layer(started); - if let Some(h) = self.class_a_hi() { - self.recent_creates.drop_ready(h); - } - } - - pub fn expire_recent_creates(&self, tip_hint: u32) { - self.expire_recent_creates_defer(tip_hint); - self.flush_recent_creates(); + /// Freeze in-flight keep-until at write (`until = lookup_started_hi.max(hi)`). + pub fn stamp_create_keep_until(&self, height: u32) { + let until = self.lookup_started_hi().unwrap_or(0).max(height); + let mut g = self + .create_keep_until + .lock() + .unwrap_or_else(|p| p.into_inner()); + g.entry(height) + .and_modify(|u| *u = (*u).max(until)) + .or_insert(until); } - pub fn expire_recent_creates_defer(&self, _tip_hint: u32) { - if let Some(h) = self.class_a_hi() { - self.recent_creates.drop_ready(h); - } + /// Load prune: apply then clear pending keep-until stamps. + pub fn take_create_keep_until(&self) -> Vec<(u32, u32)> { + let mut g = self + .create_keep_until + .lock() + .unwrap_or_else(|p| p.into_inner()); + std::mem::take(&mut *g).into_iter().collect() } /// Index-only queue entries (no payload clone). Empty after restart. @@ -1924,7 +1856,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 rec = self.recent_creates.size_detail(); let (union_layers, union_keys) = self.published_ids.size_snapshot(); let h2h_keys = self .height_by_hash @@ -1948,12 +1879,12 @@ impl Query { pstore_weak: mem.pstore_weak, pstore_live: mem.pstore_live, pstore_bytes: mem.pstore_bytes, - recent_heights: rec.0, - recent_keys: rec.1, - recent_pub_keys: rec.2, - recent_overlay_keys: rec.3, - recent_fifo_keys: rec.4, - recent_pin_bytes: self.recent_creates.approx_pin_bytes(), + recent_heights: 0, + recent_keys: 0, + recent_pub_keys: 0, + recent_overlay_keys: 0, + recent_fifo_keys: 0, + recent_pin_bytes: 0, union_layers, union_keys, h2h_keys, @@ -2599,20 +2530,13 @@ mod tests { } #[test] - fn flush_recent_creates_tags_until_from_lookup_started_hi() { - let (dir, q) = temp_query("flush-until"); - let mut tid = [0u8; 32]; - tid[0] = 1; + fn stamp_create_keep_until_freezes_lookup_started_hi() { + let (dir, q) = temp_query("keep-until"); q.set_lookup_started_hi(Some(40)); - q.note_recent_creates_rows(10, [(tid, Fk(1), (1, 2))]); - q.flush_recent_creates(); - assert_eq!(q.recent_creates().get(&tid), Some((Fk(1), (1, 2)))); - q.set_class_a_hi(Some(39)); - q.flush_recent_creates(); - assert_eq!(q.recent_creates().get(&tid), Some((Fk(1), (1, 2)))); - q.set_class_a_hi(Some(40)); - q.flush_recent_creates(); - assert!(q.recent_creates().get(&tid).is_none()); + q.stamp_create_keep_until(10); + let rows = q.take_create_keep_until(); + assert_eq!(rows, vec![(10, 40)]); + assert!(q.take_create_keep_until().is_empty()); let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-query/src/recent_creates.rs b/crates/rbitcoin-query/src/recent_creates.rs deleted file mode 100644 index f5478c61..00000000 --- a/crates/rbitcoin-query/src/recent_creates.rs +++ /dev/null @@ -1,596 +0,0 @@ -//! Write-published identity + optional [`CreatePin`] as an Arc layer chain. -//! -//! Same splice/prepend as live_union ([`crate::layer_chain`]). One layer per -//! [`RecentCreates::publish_layer`]. Load snapshot walks pending then the head. -//! [`RecentSnap::get`] is fk+range; [`RecentSnap::create_pin`] clones the pin Arc. - -use crate::layer_chain::{self, ChainLayer}; -use crate::published_ids::TxidHasher; -use crate::CreatePin; -use arc_swap::ArcSwapOption; -use rbitcoin_primitives::Fk; -use std::collections::HashMap; -use std::hash::BuildHasherDefault; -use std::sync::{Arc, Mutex, OnceLock}; - -type LiveMap = HashMap<[u8; 32], LiveEnt, BuildHasherDefault>; -type RecentLayer = ChainLayer; - -fn empty_pending() -> Arc { - static EMPTY: OnceLock> = OnceLock::new(); - Arc::clone(EMPTY.get_or_init(|| Arc::new(LiveMap::default()))) -} - -#[derive(Clone)] -struct LiveEnt { - fk: Fk, - range: (u64, u64), - height: u32, - outs: Option, -} - -struct Inner { - pending: Arc, -} - -/// Pending notes plus the published layer head. -#[derive(Clone)] -pub struct RecentSnap { - head: Option>, - pending: std::sync::Arc, -} - -impl RecentSnap { - fn ent(&self, txid: &[u8; 32]) -> Option<(Fk, (u64, u64), Option)> { - if *txid == [0u8; 32] { - return None; - } - if let Some(e) = self.pending.get(txid) { - return Some((e.fk, e.range, e.outs.clone())); - } - self.head.as_ref()?.walk(|layer| { - layer - .hits - .get(txid) - .map(|e| (e.fk, e.range, e.outs.clone())) - }) - } - - pub fn get(&self, txid: &[u8; 32]) -> Option<(Fk, (u64, u64))> { - self.ent(txid).map(|(fk, range, _)| (fk, range)) - } - - /// Arc-clone the create pin when the note carried one. Identity-only notes - /// return `None` (load pin then cold-fills). - pub fn create_pin(&self, txid: &[u8; 32]) -> Option { - self.ent(txid).and_then(|(_, _, outs)| outs) - } -} - -/// Write-published, load-read identity ring. -pub struct RecentCreates { - head: ArcSwapOption, - inner: Mutex, -} - -impl Default for RecentCreates { - fn default() -> Self { - Self { - head: ArcSwapOption::empty(), - inner: Mutex::new(Inner { - pending: empty_pending(), - }), - } - } -} - -impl RecentCreates { - pub fn new() -> Self { - Self::default() - } - - /// Freeze pending into one layer and prepend. No-op when pending is empty. - /// Does not clone older `hits` Arcs. - pub fn publish_layer(&self, until: u32) { - let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - if g.pending.is_empty() { - return; - } - let hits = std::mem::replace(&mut g.pending, empty_pending()); - drop(g); - let mut lo = u32::MAX; - let mut hi = 0u32; - for e in hits.values() { - lo = lo.min(e.height); - hi = hi.max(e.height); - } - let until = until.max(hi); - layer_chain::rcu_head(&self.head, |older| { - Some(ChainLayer::prepend( - older.clone(), - lo, - hi, - until, - Arc::clone(&hits), - )) - }); - } - - /// Merge pending into a layer that is never drop_ready until Class A HWMs exist. - pub fn publish_if_dirty(&self) { - self.publish_layer(u32::MAX); - } - - /// Drop layers with `until <= class_a_hi`. Kept nodes reuse `hits` Arc. - pub fn drop_ready(&self, class_a_hi: u32) { - layer_chain::rcu_head(&self.head, |head| { - layer_chain::splice_kept(head.clone(), |l| l.meta > class_a_hi) - }); - } - - pub fn note(&self, height: u32, rows: impl IntoIterator) { - self.note_pins(height, rows.into_iter().map(|(t, f, r)| (t, f, r, None))); - } - - pub fn note_pins( - &self, - height: u32, - rows: impl IntoIterator)>, - ) { - let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - let pending = Arc::make_mut(&mut g.pending); - for (txid, fk, range, outs) in rows { - if txid == [0u8; 32] { - continue; - } - pending.insert( - txid, - LiveEnt { - fk, - range, - height, - outs, - }, - ); - } - } - - /// Drop published keys / pending with `LiveEnt.height ≤ through`. - pub fn expire_through(&self, through: u32) { - self.filter_pending(|e| e.height > through); - self.filter_layers(|l| l.hi > through, |e| e.height > through); - } - - /// Disconnect: drop heights `≥ height`. - pub fn drop_from(&self, height: u32) { - self.filter_pending(|e| e.height < height); - self.filter_layers(|l| l.lo < height, |e| e.height < height); - } - - fn filter_pending(&self, keep: impl Fn(&LiveEnt) -> bool) { - let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - if g.pending.is_empty() { - return; - } - Arc::make_mut(&mut g.pending).retain(|_, e| keep(e)); - if g.pending.is_empty() { - g.pending = empty_pending(); - } - } - - fn filter_layers( - &self, - keep_layer: impl Fn(&RecentLayer) -> bool, - keep_ent: impl Fn(&LiveEnt) -> bool, - ) { - layer_chain::rcu_head(&self.head, |head| { - filter_chain(head.clone(), &keep_layer, &keep_ent) - }); - } - - pub fn get(&self, txid: &[u8; 32]) -> Option<(Fk, (u64, u64))> { - self.snapshot().get(txid) - } - - pub fn create_pin(&self, txid: &[u8; 32]) -> Option { - self.snapshot().create_pin(txid) - } - - pub fn snapshot(&self) -> RecentSnap { - let g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - RecentSnap { - head: self.head.load_full(), - pending: Arc::clone(&g.pending), - } - } - - #[cfg(test)] - fn head_hits_arc(&self) -> Option> { - self.head.load_full().map(|h| Arc::clone(&h.hits)) - } - - /// Occupancy for `ibd: sizes`. - pub fn size_snapshot(&self) -> (usize, usize) { - let d = self.size_detail(); - (d.0, d.1) - } - - /// `(layers, live_keys, pub_keys, pending_keys, live_keys)`. - pub fn size_detail(&self) -> (usize, usize, usize, usize, usize) { - let g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - let pending = g.pending.len(); - drop(g); - let mut layers = 0usize; - let mut pub_k = 0usize; - let mut cur = self.head.load_full(); - while let Some(layer) = cur { - layers = layers.saturating_add(1); - pub_k = pub_k.saturating_add(layer.hits.len()); - cur = layer.older.clone(); - } - let live = pub_k.saturating_add(pending); - (layers, live, pub_k, pending, live) - } - - pub fn approx_pin_bytes(&self) -> u64 { - let g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - let mut n = 0u64; - for e in g.pending.values() { - if let Some(p) = &e.outs { - n = n.saturating_add(crate::archive::create_pin_approx_bytes(p) as u64); - } - } - drop(g); - let mut cur = self.head.load_full(); - while let Some(layer) = cur { - for e in layer.hits.values() { - if let Some(p) = &e.outs { - n = n.saturating_add(crate::archive::create_pin_approx_bytes(p) as u64); - } - } - cur = layer.older.clone(); - } - n - } -} - -fn layer_ents_all(l: &RecentLayer, keep: impl Fn(&LiveEnt) -> bool) -> bool { - l.hits.values().all(keep) -} - -fn layer_ents_any(l: &RecentLayer, keep: impl Fn(&LiveEnt) -> bool) -> bool { - l.hits.values().any(keep) -} - -fn filter_chain( - head: Option>, - keep_layer: &impl Fn(&RecentLayer) -> bool, - keep_ent: &impl Fn(&LiveEnt) -> bool, -) -> Option> { - let mut nodes = Vec::new(); - let mut cur = head; - while let Some(n) = cur { - let older = n.older.clone(); - nodes.push(n); - cur = older; - } - let mut new_head: Option> = None; - for n in nodes.into_iter().rev() { - if keep_layer(&n) && layer_ents_all(&n, keep_ent) { - new_head = Some(ChainLayer::prepend( - new_head, - n.lo, - n.hi, - n.meta, - Arc::clone(&n.hits), - )); - continue; - } - if !layer_ents_any(&n, keep_ent) { - continue; - } - let mut hits = LiveMap::default(); - let mut lo = u32::MAX; - let mut hi = 0u32; - for (k, e) in n.hits.iter() { - if keep_ent(e) { - lo = lo.min(e.height); - hi = hi.max(e.height); - hits.insert(*k, e.clone()); - } - } - if hits.is_empty() { - continue; - } - new_head = Some(ChainLayer::prepend( - new_head, - lo, - hi, - n.meta, - Arc::new(hits), - )); - } - new_head -} - -#[cfg(test)] -mod tests { - use super::*; - - fn tid(b: u8) -> [u8; 32] { - let mut t = [0u8; 32]; - t[0] = b; - t - } - - #[test] - fn recent_layer_publish_prepends_without_cloning_older_hits() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.publish_layer(100); - let first = r.head_hits_arc().expect("head"); - r.note(11, [(tid(2), Fk(2), (3, 4))]); - r.publish_layer(101); - let older = r - .head - .load_full() - .expect("head") - .older - .clone() - .expect("older"); - assert!( - Arc::ptr_eq(&older.hits, &first), - "second publish must prepend, not clone the older hits Arc" - ); - assert_eq!(r.get(&tid(1)), Some((Fk(1), (1, 2)))); - assert_eq!(r.get(&tid(2)), Some((Fk(2), (3, 4)))); - } - - #[test] - fn recent_layer_drop_ready_splices_older_keeps_newer_hits_arc() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.publish_layer(10); - r.note(11, [(tid(2), Fk(2), (3, 4))]); - r.publish_layer(40); - let newer = r.head_hits_arc().expect("newer"); - r.drop_ready(10); - assert!(r.get(&tid(1)).is_none(), "until=10 must drop at class_a 10"); - assert_eq!(r.get(&tid(2)), Some((Fk(2), (3, 4)))); - assert!( - Arc::ptr_eq(&r.head_hits_arc().expect("kept"), &newer), - "kept layer hits Arc must not be cloned on splice" - ); - } - - #[test] - fn two_notes_without_flush_do_not_rebuild_snapshot() { - let r = RecentCreates::new(); - let before = r.snapshot(); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.note(11, [(tid(2), Fk(2), (3, 4))]); - let mid = r.snapshot(); - assert!( - match (&before.head, &mid.head) { - (None, None) => true, - (Some(a), Some(b)) => Arc::ptr_eq(a, b), - _ => false, - }, - "note must not publish a layer; publish_layer is the prepend" - ); - assert_eq!( - r.get(&tid(1)), - Some((Fk(1), (1, 2))), - "pending must serve get before publish_layer" - ); - assert_eq!(mid.get(&tid(2)), Some((Fk(2), (3, 4)))); - r.publish_if_dirty(); - assert_eq!(r.get(&tid(1)), Some((Fk(1), (1, 2)))); - assert_eq!(r.get(&tid(2)), Some((Fk(2), (3, 4)))); - let after = r.snapshot(); - assert!(after.head.is_some()); - r.publish_if_dirty(); - assert!( - Arc::ptr_eq( - after.head.as_ref().unwrap(), - r.snapshot().head.as_ref().unwrap() - ), - "second publish_if_dirty is a no-op when pending is empty" - ); - } - - #[test] - fn note_makes_get_visible() { - let r = RecentCreates::new(); - assert!(r.get(&tid(1)).is_none()); - r.note(10, [(tid(1), Fk(7), (100, 8))]); - assert_eq!(r.get(&tid(1)), Some((Fk(7), (100, 8)))); - assert!(r.get(&tid(2)).is_none()); - } - - #[test] - fn zero_txid_is_never_a_hit() { - let r = RecentCreates::new(); - r.note(1, [([0u8; 32], Fk(1), (0, 1))]); - assert!(r.get(&[0u8; 32]).is_none()); - assert_eq!(r.size_snapshot(), (0, 0)); - } - - #[test] - fn expire_through_drops_old_keeps_newer() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(1), (1, 2)), (tid(2), Fk(2), (3, 4))]); - r.note(11, [(tid(2), Fk(2), (3, 4)), (tid(3), Fk(3), (5, 6))]); - r.expire_through(10); - assert!(r.get(&tid(1)).is_none(), "height-10-only key must drop"); - assert_eq!( - r.get(&tid(2)), - Some((Fk(2), (3, 4))), - "re-noted at 11 must survive expire of 10" - ); - assert_eq!(r.get(&tid(3)), Some((Fk(3), (5, 6)))); - let (_layers, keys) = r.size_snapshot(); - assert_eq!(keys, 2); - } - - #[test] - fn drop_from_removes_disconnect_height_and_above() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.note(12, [(tid(2), Fk(2), (3, 4))]); - r.drop_from(12); - assert_eq!(r.get(&tid(1)), Some((Fk(1), (1, 2)))); - assert!(r.get(&tid(2)).is_none()); - } - - fn dummy_pin(script: Vec) -> CreatePin { - Arc::new(( - rbitcoin_store::TxRecord { - txid: tid(1), - version: 1, - locktime: 0, - input_start_fk: Fk::NULL, - input_count: 0, - output_start_fk: Fk::NULL, - output_count: 1, - }, - vec![rbitcoin_store::OutputRecord::unspent(1, script)], - )) - } - - #[test] - fn recent_snap_get_does_not_need_outs() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(7), (100, 8))]); - let snap = r.snapshot(); - assert_eq!(snap.get(&tid(1)), Some((Fk(7), (100, 8)))); - assert!( - snap.create_pin(&tid(1)).is_none(), - "identity note must not clone a pin Arc on get; create_pin is None" - ); - assert!(r.create_pin(&tid(1)).is_none()); - } - - #[test] - fn recent_create_pin_survives_flush_and_expires() { - let r = RecentCreates::new(); - let pin = dummy_pin(vec![0x51, 0xaa, 0xbb]); - r.note_pins(10, [(tid(1), Fk(7), (100, 8), Some(Arc::clone(&pin)))]); - assert!( - Arc::ptr_eq(&r.create_pin(&tid(1)).expect("pending pin"), &pin), - "pending create_pin hits the same Arc" - ); - r.publish_if_dirty(); - let flushed = r.create_pin(&tid(1)).expect("published pin"); - assert!( - Arc::ptr_eq(&flushed, &pin), - "publish prepends an Arc of the map, not script bytes" - ); - assert_eq!(r.get(&tid(1)), Some((Fk(7), (100, 8)))); - r.expire_through(10); - r.publish_if_dirty(); - assert!(r.get(&tid(1)).is_none()); - assert!(r.create_pin(&tid(1)).is_none()); - } - - #[test] - fn recent_size_counts_pin_bytes() { - let r = RecentCreates::new(); - let script = vec![0x51; 400]; - let pin = dummy_pin(script.clone()); - r.note_pins(10, [(tid(1), Fk(7), (100, 8), Some(pin))]); - let n = r.approx_pin_bytes(); - assert!( - n >= script.len() as u64, - "size must count pin script bytes, not 96 B/key; got {n}" - ); - assert!(n > 96, "96 B/key identity estimate must not be the payload"); - } - - #[test] - fn snapshot_pending_is_cow_ptr_eq_until_note() { - let r = RecentCreates::new(); - let a = r.snapshot(); - let b = r.snapshot(); - assert!( - Arc::ptr_eq(&a.pending, &b.pending), - "two snapshots with no note/publish/drop share pending Arc" - ); - let pin = dummy_pin(vec![0x51, 0xaa]); - r.note_pins(10, [(tid(1), Fk(1), (1, 2), Some(Arc::clone(&pin)))]); - let c = r.snapshot(); - assert!( - !Arc::ptr_eq(&a.pending, &c.pending), - "note must COW a new pending map" - ); - assert_eq!(c.get(&tid(1)), Some((Fk(1), (1, 2)))); - assert!( - Arc::ptr_eq(&c.create_pin(&tid(1)).expect("pin"), &pin), - "create_pin Arc-clones the pin, not script bytes" - ); - assert!( - a.get(&tid(1)).is_none(), - "snapshot taken before note still sees the old map" - ); - assert!(a.create_pin(&tid(1)).is_none()); - } - - #[test] - fn empty_pending_reuses_one_empty_arc() { - let r = RecentCreates::new(); - let a = r.snapshot(); - assert!(a.pending.is_empty()); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.publish_layer(10); - let b = r.snapshot(); - assert!( - Arc::ptr_eq(&a.pending, &b.pending), - "empty pending after publish reuses the construction empty Arc" - ); - r.drop_from(10); - let c = r.snapshot(); - assert!( - Arc::ptr_eq(&a.pending, &c.pending), - "empty pending after drop_from reuses the same empty Arc" - ); - } - - #[test] - fn stale_publish_cas_does_not_restore_drop_from() { - let r = RecentCreates::new(); - r.note(10, [(tid(1), Fk(1), (1, 2))]); - r.publish_layer(10); - let stale = r.head.load_full(); - r.drop_from(10); - assert!(r.get(&tid(1)).is_none()); - r.note(12, [(tid(2), Fk(2), (3, 4))]); - let hits = { - let g = r.inner.lock().unwrap_or_else(|p| p.into_inner()); - Arc::clone(&g.pending) - }; - let mut lo = u32::MAX; - let mut hi = 0u32; - for e in hits.values() { - lo = lo.min(e.height); - hi = hi.max(e.height); - } - let next = ChainLayer::prepend(stale.clone(), lo, hi, hi.max(12), hits); - assert!( - layer_chain::cas_head(&r.head, &stale, Some(next)).is_err(), - "CAS against a stale head must fail" - ); - assert!( - r.get(&tid(1)).is_none(), - "stale publish must not restore drop_from heights" - ); - assert_eq!(r.get(&tid(2)), Some((Fk(2), (3, 4)))); - r.publish_layer(12); - assert!(r.get(&tid(1)).is_none()); - assert_eq!(r.get(&tid(2)), Some((Fk(2), (3, 4)))); - let head = r.head.load_full().expect("published"); - assert!( - head.older.is_none(), - "CAS retry must prepend onto the current head, not the stale drop_from chain" - ); - } -} diff --git a/crates/rbitcoin-query/src/stamp.rs b/crates/rbitcoin-query/src/stamp.rs index c1832456..aa6bfe2b 100644 --- a/crates/rbitcoin-query/src/stamp.rs +++ b/crates/rbitcoin-query/src/stamp.rs @@ -1,10 +1,10 @@ -//! External parent create_fk stamp: in-flight → published → recent creates → TipOnly. +//! External parent create_fk stamp: in-flight → published → TipOnly. //! //! One function for S0 plan (`archive_plan_batch_from_store`) and plan=None -//! rehydrate. Pipeline parent store is outs only — not a create_fk source. +//! rehydrate. In-flight holds CreatePins through write keep-until. use crate::published_ids::TxidHasher; -use crate::{CreatePin, InFlightView, PublishedIds, QueryError, RecentCreates, U64Map, U64Set}; +use crate::{CreatePin, InFlightView, PublishedIds, QueryError, U64Map, U64Set}; use rbitcoin_primitives::Fk; use rbitcoin_store::Store; use std::collections::HashMap; @@ -71,7 +71,7 @@ impl ExternalParentStamp { } } -/// Bind `need` txids: in-flight → published `live_union` → recent creates → leftover. +/// Bind `need` txids: in-flight → published `live_union` → leftover TipOnly. /// /// Then idx range-fill for resolved creates that have no range and no /// in-flight outs. Same-batch identities are not inputs — callers skip them @@ -81,10 +81,8 @@ pub fn stamp_external_parents( need: &[[u8; 32]], in_flight: &InFlightView, published: &PublishedIds, - recent: &RecentCreates, ) -> Result { let pub_head = published.load(); - let recent_snap = recent.snapshot(); let mut stamp = ExternalParentStamp { resolved: TxidFkMap::with_capacity_and_hasher(need.len() / 2, Default::default()), idents: U64Map::with_capacity_and_hasher(need.len(), Default::default()), @@ -100,7 +98,10 @@ pub fn stamp_external_parents( if let Some(fk) = in_flight.get_create_fk(t) { stamp.resolved.insert(*t, fk); if let Some(id) = fk.get() { - stamp.bind(id, *t); + let e = stamp.bind(id, *t); + if let Some(pin) = in_flight.get_out(id) { + e.pin = Some(std::sync::Arc::clone(pin)); + } } } else { still_need.push(t); @@ -123,24 +124,7 @@ pub fn stamp_external_parents( } stamp.pin_txid_ns = t_pin_txid.elapsed().as_nanos() as u64; - let t_recent = Instant::now(); - let mut need_head: Vec<[u8; 32]> = Vec::new(); - for t in after_pub { - if let Some((fk, range)) = recent_snap.get(t) { - stamp.resolved.insert(*t, fk); - if let Some(id) = fk.get() { - let e = stamp.bind(id, *t); - e.body = Some(range); - if let Some(pin) = recent_snap.create_pin(t) { - e.pin = Some(pin); - } - } - stamp.recent_n = stamp.recent_n.saturating_add(1); - continue; - } - need_head.push(*t); - } - stamp.recent_ns = t_recent.elapsed().as_nanos() as u64; + let mut need_head: Vec<[u8; 32]> = after_pub.into_iter().copied().collect(); stamp.head_need_n = need_head.len() as u64; let t_head = Instant::now(); diff --git a/crates/rbitcoin-test/tests/scenarios.rs b/crates/rbitcoin-test/tests/scenarios.rs index 14363a18..65378469 100644 --- a/crates/rbitcoin-test/tests/scenarios.rs +++ b/crates/rbitcoin-test/tests/scenarios.rs @@ -2039,7 +2039,6 @@ fn wire_prep_ahead_cross_batch_spend_fills_parent_layout() { parent_hash: None, next_tx_start: q.tx_body_count().saturating_add(1).max(1), in_flight: rbitcoin_query::InFlightView::empty(), - parent_store: None, published: std::sync::Arc::new(rbitcoin_query::PublishedIds::new()), }; let mat_a = confirm_wire_load_phase_pipelined( diff --git a/docs/concurrency.md b/docs/concurrency.md index c9aad5bb..49f9358d 100644 --- a/docs/concurrency.md +++ b/docs/concurrency.md @@ -31,7 +31,7 @@ only when no foreground wave and no detached job are waiting. Process restart le **Body queue:** process-local **in-RAM** FIFO (id / height / hash / header_fk / raw charge) plus a first-wins **height → id** map. Height APIs (`is_resolve_complete`, `get_by_height`, `hash_at_height`, pack snapshot) do not walk `index.values()`. **Why RAM:** avoid **double disk write** of every block (queue then Class A); accept **redownload on restart** and peak RAM of soft depth. Peer copies wire **then** locks and enqueues **raw**; lookup decodes and parks resolved, then **dequeues** on load-batch send into **loadq** (`LoadBatch`). Load stamp consumes that `pres` — it does **not** read `block_queue_resolved`. **Never both** raw and decoded. **Primary capacity is soft densify assign** (no hysteresis): under ~100 MiB free densify ahead; over ~100 MiB only heights confirm will consume in the next ~1 min at tip rate; at/over 1 GiB assign-stop (`RBITCOIN_BLOCK_QUEUE_GB` / `_BYTES`, `0` = unlimited) fill holes through the already-fetched height horizon only. Soft assign keys off raw `bytes()`. Enqueue is never byte-capped. Height horizon (`CONTIG_DENSIFY_AHEAD`, 64 k past tip) caps densify/receive walk. **Offer** on peer Block → RAM; lookup takes; write **dequeue** is a no-op if the row is already gone. Restart starts empty (legacy `store/block_queue/` is best-effort removed). -**Pipeline pins:** plan `batch_pin` / `BatchParents` only (no process create FIFO). Stamp staging (`external_parents`) is **frozen/cleared after pin** (`ArchiveWritePlan::freeze_after_pin`) so write batch concatenates `batch_pin` / `planned_fks` / SpendEdges (IBD packed ins are empty; write fills from wire). Outs live in `BatchParents` / pstore — not a third plan-local outs map. `SharedParentPin` vacant insert stores Frozen outs/layout halves; first real compose promotes that half to `ArcSwap` (RCU; no-op cover stays Frozen; no in-place mutation). `BatchParents` sticky-caches the last outs Arc for multi-input assemble. `get_parent_txout_parts` holds that Arc and yields `&[u8]`; IBD assemble below milestone counts sigops from the borrow (no `ScriptBuf::from_bytes`); scripts-on still clones into jobs. Pack `pending_spent` is `OutPointSet` (folds every hasher write — `TxidHasher` would drop the txid when vout arrives). IBD `pin(... adopt= range_fill= contract= publish=)` names residual pin wall. ConfirmParentCache holds tip-ahead **Arc** header plans only (insert/replace/drop under tip GC). **RecentCreates** is write-published identity (`txid → fk+range`) **and** full create outs (`CreatePin` Arc, same as `batch_pin`), one Arc layer per Class A write (shared splice with live_union). `until = lookup_started_hi.max(hi)` at publish; drop when `class_a_hi >= until`. Stamp carries CreatePin; pin does not re-walk. Pending notes are visible before prepend. live_union keep is BQ or `(tip, taken_hi]` only. IBD load passes **no** process `PipelineParentStore` (tip-follow may still hold one). Not a coins cache / spend FIFO. +**Pipeline pins:** plan `batch_pin` / `BatchParents` only (no process create FIFO). Stamp staging (`external_parents`) is **frozen/cleared after pin** (`ArchiveWritePlan::freeze_after_pin`) so write batch concatenates `batch_pin` / `planned_fks` / SpendEdges (IBD packed ins are empty; write fills from wire). Outs live in `BatchParents` — not a third plan-local outs map. `SharedParentPin` vacant insert stores Frozen outs/layout halves; first real compose promotes that half to `ArcSwap` (RCU; no-op cover stays Frozen; no in-place mutation). `BatchParents` sticky-caches the last outs Arc for multi-input assemble. `get_parent_txout_parts` holds that Arc and yields `&[u8]`; IBD assemble below milestone counts sigops from the borrow (no `ScriptBuf::from_bytes`); scripts-on still clones into jobs. Pack `pending_spent` is `OutPointSet` (folds every hasher write — `TxidHasher` would drop the txid when vout arrives). IBD `pin(... range_fill= contract=)` names residual pin wall. ConfirmParentCache holds tip-ahead **Arc** header plans only (insert/replace/drop under tip GC). In-flight CreatePin layers drop at drain+fence **and**, for pin layers, write-stamped `until = lookup_started_hi.max(hi)` then `class_a_hi >= until`. Stamp carries CreatePin from `InFlightView`. live_union keep is BQ or `(tip, taken_hi]` only. Not a coins cache / spend FIFO. **tx.head (segmented):** see [`heads.md`](./heads.md). Lookup: live pin by txid → hot (open + ages ≤3) → ID/idx → cold (ages ≥4) if needed. diff --git a/docs/ibd-memory.md b/docs/ibd-memory.md index 88d82e3d..46b7aac9 100644 --- a/docs/ibd-memory.md +++ b/docs/ibd-memory.md @@ -33,8 +33,8 @@ pres and **not** the raw bytes. Reorg gather that wants wire re-encodes. | Structure | Cap / bound | Production clear / evict | |-----------|-------------|---------------------------| | **Published identity union** | Per-wave map of **this wave's** parent identities (re-home union hits + TipOnly misses; `ArcSwap` of the layer-chain head; get walks, no union rebuild) | Lookup keeps a layer while its span is on the BQ or overlaps `(tip, taken_hi]`. Disconnect stores `None`. Not a process FIFO. | -| **Pipeline pins (no process FIFO)** | Plan `batch_pin` / `BatchParents` only. IBD has **no** process `PipelineParentStore` | Drop with batch. Cold **outs** for ancient parents use `txout.body` into `BatchParents` (stamped range). Recent-window first spend uses stamp-carried RecentCreates outs | -| **RecentCreates create-outs ring** | identity + full create outs (`CreatePin` Arc); one Arc layer per write phase; drop when `class_a_hi >= layer.until` (`until = lookup_started_hi.max(hi)` at publish) | Write notes `batch_pin` then one prepend+splice after Class A+idx; disconnect `drop_from`. Sizes: `recent=` counts layers/keys + CreatePin bytes. Not a coins cache / spend 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 write keep-until | +| **In-flight CreatePin layers** | identity + full create outs; pin layers drop when drain+fence **and** `class_a_hi >= until` (`until = lookup_started_hi.max(hi)` frozen at write) | Lookup/load note; write stamps keep-until; disconnect `drop_from`. Sizes: `iflight=`. Not a coins cache / spend FIFO | | **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** | on disk after post-IBD Class A recollect | bulk FullCold / ColdResume at tip; not during Direct confirm | @@ -111,7 +111,7 @@ known retain structures: | `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= pstore= recent= union= h2h= fence= fuse8= mphf_g= open_keys= class_c_l2= accounted= residual=` | Approx process heap: BQ + load-ahead CreatePins + **pstore 0 on IBD** (tip-follow may still hold a store) + **RecentCreates identity+outs ring** (`recent=Nh live=/pub=/ov= fifo=≈NMiB` from CreatePin bytes) + **PublishedIds/LiveUnion layers** (`union=NL/Nk`) + `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= pstore= recent= union= h2h= fence= fuse8= mphf_g= open_keys= class_c_l2= accounted= residual=` | Approx process heap: BQ + load-ahead CreatePins (`iflight=`) + **pstore/recent meters stay 0** (no process pin store, no RecentCreates ring) + **PublishedIds/LiveUnion layers** (`union=NL/Nk`) + `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 16c626eb..989ee68f 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -74,7 +74,7 @@ not head/idx. | Stage | Invariant | Soft path allowed? | |-------|-----------|--------------------| | Lookup parent stamp | Every external spent parent has create_fk + body_range (or offline in_flight CreatePin) + reverse txid. Archived parents also have `spent.idx` range on the stamp (in-flight outs skip idx) | Missing → hard Err at stamp / pin contract | -| Parent create_fk union | **in-flight** (prune **after** pin + scripts handoff: drain inserted fk span **and** `fence.covers_fk_span`) → **published live_union** (height-layered chain, `ArcSwap` of the head, get walks; no union rebuild) → **RecentCreates** (write-published Arc layers: `txid → fk+range` **and** optional `CreatePin`; drop when `class_a_hi >= until`; stamp carries CreatePin, pin does not re-walk) → **TipOnly** (connected **and** idx `body_range`; older-than-window leftover). One helper: [`stamp_external_parents`](../crates/rbitcoin-query/src/stamp.rs) (S0 plan and plan=None). No leftover pending map, no process pin FIFO, no BQ-side hits map, no parent-store create_fk on stamp. Each new layer is this wave's parent identities (re-home union hits + TipOnly misses); `skipped=` skips TipOnly IO, not the layer row. Lookup drops a layer once no height in **its span** remains on the BQ and the span does not overlap `(tip, taken_hi]`. Disconnect `store(None)` immediately and clears the lookup union and recent ring from that height. Header-cache GC polls store tip each load pack. One fk per txid — [`errata.md`](./errata.md). | **No** soft-requeue. Union miss → `Corrupt("parent create_fk unresolved")` (permanent). Identity without idx range → `Corrupt("invariant: idx range missing after identity")`, not a miss | +| Parent create_fk union | **in-flight** (prune **after** pin + scripts handoff: drain inserted fk span **and** `fence.covers_fk_span`; pin layers also wait for write-stamped `until = lookup_started_hi.max(hi)` then `class_a_hi >= until`) → **published live_union** (height-layered chain, `ArcSwap` of the head, get walks; no union rebuild) → **TipOnly** (connected **and** idx `body_range`; leftover after drain+fence). One helper: [`stamp_external_parents`](../crates/rbitcoin-query/src/stamp.rs) (S0 plan and plan=None). No leftover pending map, no process pin FIFO, no BQ-side hits map, no parent-store create_fk on stamp. Each new layer is this wave's parent identities (re-home union hits + TipOnly misses); `skipped=` skips TipOnly IO, not the layer row. Lookup drops a layer once no height in **its span** remains on the BQ and the span does not overlap `(tip, taken_hi]`. Disconnect `store(None)` immediately and clears the lookup union from that height. Header-cache GC polls store tip each load pack. One fk per txid — [`errata.md`](./errata.md). | **No** soft-requeue. Union miss → `Corrupt("parent create_fk unresolved")` (permanent). Identity without idx range → `Corrupt("invariant: idx range missing after identity")`, not a miss | | io_uring harvest | Pending `user_data` set + kind/epoch. Unexpected CQE, undrained leftover, CQ overflow, wait timeout. Held-session `begin_batch` drains leftover before a new kind/epoch wave and fails closed (`Err`) if undrained or poisoned. Drain does not ignore unmatched CQE / CQ overflow. Held idx fill must not libc-fallback on a dirty ring | **No** silent success. `Corrupt("invariant: io_uring …")` (not `bdz g page bad slot`). Poison + drop TLS ring. 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 idx IO) | **No** idx cold outs on load; **no** `spent.idx` IO on pin; **no** `inwit` on pin | | Ensure (write) | Every non-null spend edge has `spent_range` abs after ensure returns. Lookup already stamped archived parents; write `tx_spent_range_batch` only for unstamped fks (same-batch after Class A, holes) | Idx stamp of remaining `spent.body` ranges; incomplete → `invariant:` | @@ -164,7 +164,7 @@ bind. | `optimistic_assemble_unstamped_parent_is_invariant` | Optimistic assemble: pin miss is lookup invariant, not head recover | | `parent_pin_stamp_take_from_plan_moves_maps` | S0 `take_from_plan` leaves `resolved` empty (no txid→fk invert) | | `plan_batch_one_fill_missing_when_parents_already_stamped` | one `fill_missing_parent_ranges` when packed adds no new fks | -| `direct_write_skips_create_pin_map_recent_matches_idx` | Direct skips `write_create_pins`; RecentCreates body range = idx | +| `direct_write_skips_create_pin_map_idx_without_recent` | Direct skips `write_create_pins`; Class A idx holds body range | | `pin_takes_stamp_parent_vouts` / `plan_batch_same_header_vouts_skipped_cross_height_pinned` | pin takes stamp vouts; same-header creates not pinned | | `confirm_engine_pins_spend_of_just_written_pack` | IBD load: child spend of just-written pack (187 denserels miss) | | `confirm_reject_blacklist_surface` | fk mismatch / connect height not tip+1 soft requeue |