query: merge_outs no-op keeps outs Arc (Q-M3)

CHANGELOG.md

@@ -54,6 +54,10 @@ before 1.0).
 
 ### Changed
 
+- **Pin `merge_outs` no-op Arc:** empty / already-covered `checked`+`live`
+  keep the outs Arc (assemble sticky `ptr_eq`). RCU compose borrows `live`
+  instead of cloning script bytes on every retry.
+
 - **IBD Class A ins at write:** `confirm_wire_lookup_stamp` plans from
   wire (`archive_plan_batch_from_wire`) without `TxApply`. Packed ins stay
   empty; SpendEdges + CreatePin remain. Write encodes ins from `Arc<Block>`

crates/rbitcoin-query/src/batch_parents.rs

@@ -125,6 +125,10 @@ impl PinOuts {
         live.iter().all(|(v, _)| self.get(*v).is_some())
     }
 
+    fn already_covers(&self, live: &[(u32, OutputRecord)], checked: &[u32]) -> bool {
+        self.covers_need(checked) && (live.is_empty() || self.has_all_live(live))
+    }
+
     #[cfg(test)]
     fn live_len(&self) -> usize {
         match self {
@@ -144,15 +148,15 @@ impl PinOuts {
     }
 
     /// Compose wider need coverage (new half; does not mutate `self`).
-    fn compose(&self, live: Vec<(u32, OutputRecord)>, checked: &[u32]) -> Self {
+    fn compose(&self, live: &[(u32, OutputRecord)], checked: &[u32]) -> Self {
         match self {
             Self::Full { pin, checked: ch } => {
                 let extra = live.iter().any(|(v, _)| pin.1.get(*v as usize).is_none());
                 if extra {
                     let mut outs = self.sparse_live();
                     for (v, o) in live {
-                        if outs.binary_search_by_key(&v, |(dv, _)| *dv).is_err() {
-                            outs.push((v, o));
+                        if outs.binary_search_by_key(v, |(dv, _)| *dv).is_err() {
+                            outs.push((*v, o.clone()));
                         }
                     }
                     outs.sort_unstable_by_key(|(v, _)| *v);
@@ -178,8 +182,8 @@ impl PinOuts {
             Self::Sparse { outs, checked: ch } => {
                 let mut next_outs = outs.clone();
                 for (v, o) in live {
-                    if next_outs.binary_search_by_key(&v, |(dv, _)| *dv).is_err() {
-                        next_outs.push((v, o));
+                    if next_outs.binary_search_by_key(v, |(dv, _)| *dv).is_err() {
+                        next_outs.push((*v, o.clone()));
                     }
                 }
                 next_outs.sort_unstable_by_key(|(v, _)| *v);
@@ -392,19 +396,17 @@ impl SharedParentPin {
         self.load_outs().covers_need(need)
     }
 
+    /// No-op (empty / already-covered `checked`+`live`) keeps the outs Arc.
     fn merge_outs(&self, live: Vec<(u32, OutputRecord)>, checked: &[u32]) {
         let snap = self.load_outs();
-        if snap.covers_need(checked) && (live.is_empty() || snap.has_all_live(&live)) {
+        if snap.already_covers(&live, checked) {
             return;
         }
         self.outs.rcu(|cur| {
-            if !checked.is_empty()
-                && cur.covers_need(checked)
-                && (live.is_empty() || cur.has_all_live(&live))
-            {
+            if cur.already_covers(&live, checked) {
                 Arc::clone(cur)
             } else {
-                Arc::new(cur.compose(live.clone(), checked))
+                Arc::new(cur.compose(&live, checked))
             }
         });
     }
@@ -794,9 +796,7 @@ impl BatchParents {
             std::collections::hash_map::Entry::Occupied(o) => {
                 let p = o.get();
                 let outs = p.load_outs();
-                let need_outs = !checked.is_empty()
-                    && !(outs.covers_need(&checked)
-                        && (live.is_empty() || outs.has_all_live(&live)));
+                let need_outs = !outs.already_covers(&live, &checked);
                 if need_outs {
                     p.apply_pin_delta(
                         Some((live, checked.as_slice())),
@@ -845,7 +845,7 @@ impl BatchParents {
             std::collections::hash_map::Entry::Occupied(o) => {
                 let p = o.get();
                 let outs = p.load_outs();
-                let need_outs = !checked.is_empty() && !outs.covers_need(&checked);
+                let need_outs = !outs.covers_need(&checked);
                 if need_outs {
                     let live = {
                         let (_tx, rows) = pin.as_ref();
@@ -1794,6 +1794,78 @@ mod tests {
         assert!(pin.outs.rcu.get().is_none(), "no-op cover must stay Frozen");
     }
 
+    /// Q-M3: empty checked / already-covered live must not publish a new outs Arc.
+    #[test]
+    fn merge_outs_empty_checked_keeps_outs_arc() {
+        let mut bp = BatchParents::new();
+        bp.insert_owned(
+            Fk(1),
+            tx(1),
+            vec![(0, out(10))],
+            vec![0],
+            Some(false),
+            None,
+            Vec::new(),
+        );
+        let pin = Arc::clone(bp.pins.get(&1).unwrap());
+        assert!(pin.outs.rcu.get().is_none());
+        let before = pin.load_outs();
+        pin.merge_outs(vec![], &[]);
+        assert!(
+            Arc::ptr_eq(&before, &pin.load_outs()),
+            "empty checked no-op must keep outs Arc"
+        );
+        assert!(
+            pin.outs.rcu.get().is_none(),
+            "empty checked no-op must stay Frozen"
+        );
+        pin.merge_outs(vec![(0, out(10))], &[]);
+        assert!(
+            Arc::ptr_eq(&before, &pin.load_outs()),
+            "redundant live + empty checked must keep outs Arc"
+        );
+        assert!(pin.outs.rcu.get().is_none());
+
+        bp.insert_owned(
+            Fk(1),
+            tx(1),
+            vec![(1, out(20))],
+            vec![],
+            None,
+            None,
+            Vec::new(),
+        );
+        let after = pin.load_outs();
+        assert!(
+            after.covers_need(&[0]) && after.get(1).is_some(),
+            "Occupied empty-checked new live must still widen"
+        );
+        assert!(!Arc::ptr_eq(&before, &after));
+    }
+
+    #[test]
+    fn merge_outs_large_script_widens_once() {
+        let script = vec![0x51u8; 4096];
+        let mut bp = BatchParents::new();
+        bp.insert_owned(
+            Fk(1),
+            tx(1),
+            vec![(0, out(10))],
+            vec![0],
+            Some(false),
+            None,
+            Vec::new(),
+        );
+        let pin = Arc::clone(bp.pins.get(&1).unwrap());
+        let rec = OutputRecord::unspent(20, script.clone());
+        pin.merge_outs(vec![(1, rec.clone())], &[1]);
+        pin.merge_outs(vec![(1, rec)], &[1]);
+        let snap = pin.load_outs();
+        assert_eq!(snap.live_len(), 2);
+        assert_eq!(snap.get(1).expect("vout 1").script.len(), 4096);
+        assert_eq!(snap.get(1).unwrap().script, script);
+    }
+
     #[test]
     fn compose_adds_vout_without_mutating_old_snap() {
         let mut bp = BatchParents::new();

docs/algo-review.md

@@ -151,8 +151,7 @@ What the node runs. Findings follow.
 
 ### Query
 
-- **Q-M3.** `merge_outs` clones script bytes on every RCU retry; empty
-  `checked` always publishes a new Arc (breaks `ptr_eq` / sticky).
+*(none remaining)*
 
 ### Net / mempool / RPC / esplora
 
@@ -342,7 +341,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 | merge_outs clone | BQ scan, SipHash in-flight |
+| query | 0 | *(none remaining)* | 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 |