query: RecentCreates snapshot COW and head RCU

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<Meta: Clone, Hits>(
     }
     new_head
 }
+
+fn option_arc_eq<T>(a: &Option<Arc<T>>, b: &Option<Arc<T>>) -> 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<Meta, Hits>(
+    slot: &ArcSwapOption<ChainLayer<Meta, Hits>>,
+    expected: &Option<Arc<ChainLayer<Meta, Hits>>>,
+    next: Option<Arc<ChainLayer<Meta, Hits>>>,
+) -> Result<(), Option<Arc<ChainLayer<Meta, Hits>>>> {
+    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<Meta, Hits>(
+    slot: &ArcSwapOption<ChainLayer<Meta, Hits>>,
+    mut f: impl FnMut(&Option<Arc<ChainLayer<Meta, Hits>>>) -> Option<Arc<ChainLayer<Meta, Hits>>>,
+) {
+    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<ChainLayer<u32, u8>> = 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());
+    }
+}

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<TxidHasher>>;
 type RecentLayer = ChainLayer<u32, LiveMap>;
 
+fn empty_pending() -> Arc<LiveMap> {
+    static EMPTY: OnceLock<Arc<LiveMap>> = 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<LiveMap>,
 }
 
 /// 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<Item = ([u8; 32], Fk, (u64, u64))>) {
@@ -132,11 +138,12 @@ impl RecentCreates {
         rows: impl IntoIterator<Item = ([u8; 32], Fk, (u64, u64), Option<CreatePin>)>,
     ) {
         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,16 +169,23 @@ 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(
         &self,
         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"
+        );
+    }
 }

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<LiveMap>` |
 | P18 | BQ `dequeue` height remap | O(queue) `index.iter().find` | `HashMap<u32, VecDeque<id>>` |
 | P19 | `BlockCache` prefix drop | O(chain) per connect | `VecDeque` + base height |
 | P20 | GBT `depends` | O(txs × inputs) linear scan | `HashMap<Txid, idx>` |
@@ -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 |