diff --git a/crates/rbitcoin-query/src/layer_chain.rs b/crates/rbitcoin-query/src/layer_chain.rs index 9f7900ea..d2ac89a3 100644 --- a/crates/rbitcoin-query/src/layer_chain.rs +++ b/crates/rbitcoin-query/src/layer_chain.rs @@ -1,5 +1,6 @@ //! Newest-first Arc layer list. live_union and RecentCreates share splice/prepend. +use arc_swap::{ArcSwapOption, Guard}; use std::sync::Arc; /// One immutable layer plus the older chain. `hits` is never cloned on splice. @@ -81,3 +82,68 @@ 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/recent_creates.rs b/crates/rbitcoin-query/src/recent_creates.rs index 08daf795..f5478c61 100644 --- a/crates/rbitcoin-query/src/recent_creates.rs +++ b/crates/rbitcoin-query/src/recent_creates.rs @@ -11,11 +11,16 @@ use arc_swap::ArcSwapOption; use rbitcoin_primitives::Fk; use std::collections::HashMap; use std::hash::BuildHasherDefault; -use std::sync::{Arc, Mutex}; +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, @@ -25,7 +30,7 @@ struct LiveEnt { } struct Inner { - pending: LiveMap, + pending: Arc, } /// Pending notes plus the published layer head. @@ -73,7 +78,7 @@ impl Default for RecentCreates { Self { head: ArcSwapOption::empty(), inner: Mutex::new(Inner { - pending: LiveMap::default(), + pending: empty_pending(), }), } } @@ -91,23 +96,24 @@ impl RecentCreates { if g.pending.is_empty() { return; } - let hits = std::mem::take(&mut g.pending); + 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); } - drop(g); let until = until.max(hi); - let older = self.head.load_full(); - self.head.store(Some(ChainLayer::prepend( - older, - lo, - hi, - until, - Arc::new(hits), - ))); + 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. @@ -117,9 +123,9 @@ impl RecentCreates { /// Drop layers with `until <= class_a_hi`. Kept nodes reuse `hits` Arc. pub fn drop_ready(&self, class_a_hi: u32) { - let head = self.head.load_full(); - let next = layer_chain::splice_kept(head, |l| l.meta > class_a_hi); - self.head.store(next); + 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) { @@ -132,11 +138,12 @@ impl RecentCreates { 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; } - g.pending.insert( + pending.insert( txid, LiveEnt { fk, @@ -162,7 +169,13 @@ impl RecentCreates { fn filter_pending(&self, keep: impl Fn(&LiveEnt) -> bool) { let mut g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); - g.pending.retain(|_, e| keep(e)); + 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( @@ -170,8 +183,9 @@ impl RecentCreates { keep_layer: impl Fn(&RecentLayer) -> bool, keep_ent: impl Fn(&LiveEnt) -> bool, ) { - let head = self.head.load_full(); - self.head.store(filter_chain(head, &keep_layer, &keep_ent)); + 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))> { @@ -186,7 +200,7 @@ impl RecentCreates { let g = self.inner.lock().unwrap_or_else(|p| p.into_inner()); RecentSnap { head: self.head.load_full(), - pending: Arc::new(g.pending.clone()), + pending: Arc::clone(&g.pending), } } @@ -492,4 +506,91 @@ mod tests { ); 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/docs/algo-review.md b/docs/algo-review.md index 528179d9..f6d04405 100644 --- a/docs/algo-review.md +++ b/docs/algo-review.md @@ -48,7 +48,7 @@ What the node runs. Findings follow. |-------|-----------|--------| | Stamp | in-flight → live_union → RecentCreates → TipOnly | miss is permanent | | `LiveUnion` | ArcSwap chain of identity layers | get walks; splice keep by BQ / taken / horizon | -| `RecentCreates` | height-FIFO + ArcSwap head + pending overlay | write-published; load-then-store (sole mutator unenforced) | +| `RecentCreates` | Arc layer chain + COW pending | write-published; head CAS/RCU | | `SharedParentPin` | Frozen `OnceLock` → `ArcSwap` RCU | compose-only; no in-place mutation | | `PinOuts` / `ParentLayout` | sorted sparse vecs, `binary_search` | `covers_need` = examined, not necessarily live | | `TxidHasher` | last 8-byte write wins | sound only for bare `[u8;32]` | @@ -154,9 +154,6 @@ What the node runs. Findings follow. - **Q-M1.** `retain_headers_needing_body` (`archive.rs`): missing `first` fk → `unwrap_or(0)` keeps the wrong span. Should be `Corrupt("invariant: …")`. -- **Q-M2.** `RecentCreates` head is load-then-store, not RCU - (`recent_creates.rs`). Safe only if the write thread is the sole mutator - — unenforced. `drop_from` is on the reorg path. - **Q-M3.** `merge_outs` clones script bytes on every RCU retry; empty `checked` always publishes a new Arc (breaks `ptr_eq` / sticky). @@ -225,7 +222,6 @@ Grouped by cost at mainnet scale. Many overlap §3. | P13 | `find_free_slot` | O(cap) per admit, O(cap²) fill | free-list `VecDeque` | | P14 | `persist_all` | rewrite entire mempool files | append body + patch slot | | P15 | `evict_nonfinal` | O(n²) per reorg | worklist of affected txs | -| P17 | `RecentCreates::snapshot` | clone pending map (pins) / wave | COW `Arc` | | P18 | BQ `dequeue` height remap | O(queue) `index.iter().find` | `HashMap>` | | P19 | `BlockCache` prefix drop | O(chain) per connect | `VecDeque` + base height | | P20 | GBT `depends` | O(txs × inputs) linear scan | `HashMap` | @@ -349,7 +345,7 @@ Intentional COMPAT Electrum status extra field is **not** counted as High. |-------|------|--------|------------------| | store | 0 | seqlock, flush lost-update, fuse8 OOB, spender cycle, sidecar fsync, runs_io | BDZ fill, bulk_fill, SH N² | | consensus + primitives | 0 | *(none remaining)* | historical MTP walks, rehash txids | -| query | 0 | retain fallback, RecentCreates CAS, merge_outs clone | snapshot clone, BQ scan, SipHash in-flight | +| query | 0 | retain fallback, merge_outs clone | BQ scan, SipHash in-flight | | net | 0 | compact indexes, random eviction, AddrMan/cmpct unbounded, v2 copies | INV flush, BlockCache | | mempool | 0 | orphan vout, eviction tie, persist order, package feerate | free slot, persist_all | | rpc | 0 | submitblock gate, gettxout, hashps, unbounded batch, blockmintxfee, maxfeerate | GBT depends, longpoll |