mempool: fixes

crates/brk_mempool/src/cpfp.rs

@@ -0,0 +1,108 @@
+//! CPFP (Child Pays For Parent) cluster reasoning for live mempool
+//! transactions. Cluster scope is the seed's projected block: txs in
+//! other projected blocks share no mining fate with the seed, so
+//! including them in `effectiveFeePerVsize` would be misleading.
+//!
+//! Confirmed-tx CPFP (the same-block connected component on the
+//! chain) lives in `brk_query`, since it reads indexer/computer vecs.
+
+use brk_types::{CpfpEntry, CpfpInfo, FeeRate, Sats, TxidPrefix, VSize, Weight};
+use rustc_hash::FxHashSet;
+
+use crate::{Mempool, TxEntry};
+
+/// Cap matches Bitcoin Core's default mempool ancestor/descendant
+/// chain limits and `confirmed_cpfp`'s cap.
+const MAX: usize = 25;
+
+impl Mempool {
+    /// CPFP info for a live mempool tx, scoped to the seed's projected
+    /// block. Returns `None` if the tx is not in the mempool, so
+    /// callers can fall through to the confirmed path. Returns `Some`
+    /// with empty arms if the tx is in the mempool but below the
+    /// projection floor (no projected block to share fate with).
+    pub fn cpfp_info(&self, prefix: &TxidPrefix) -> Option<CpfpInfo> {
+        let snapshot = self.snapshot();
+        let entries = self.entries();
+        let seed_idx = entries.idx_of(prefix)?;
+        let seed = entries.slot(seed_idx)?;
+
+        let mut sum_fee = u64::from(seed.fee);
+        let mut sum_vsize = u64::from(seed.vsize);
+        let mut ancestors: Vec<CpfpEntry> = Vec::new();
+        let mut descendants: Vec<CpfpEntry> = Vec::new();
+
+        if let Some(seed_block) = snapshot.block_of(seed_idx) {
+            // Ancestor BFS gated to the seed's projected block.
+            // `visited` dedupes the walk; stale parent prefixes
+            // (confirmed/evicted between snapshot and now) are skipped
+            // when `idx_of` returns None.
+            let mut visited: FxHashSet<TxidPrefix> = FxHashSet::default();
+            visited.insert(*prefix);
+            let mut stack: Vec<TxidPrefix> = seed.depends.iter().copied().collect();
+            while let Some(p) = stack.pop() {
+                if ancestors.len() >= MAX {
+                    break;
+                }
+                if !visited.insert(p) {
+                    continue;
+                }
+                let Some(idx) = entries.idx_of(&p) else { continue };
+                if snapshot.block_of(idx) != Some(seed_block) {
+                    continue;
+                }
+                let Some(anc) = entries.slot(idx) else { continue };
+                sum_fee += u64::from(anc.fee);
+                sum_vsize += u64::from(anc.vsize);
+                ancestors.push(to_entry(anc));
+                stack.extend(anc.depends.iter().copied());
+            }
+
+            // Descendant sweep. `desc_set` starts with only the seed
+            // so siblings (txs sharing an ancestor with seed but not
+            // downstream of it) are excluded. The topological ordering
+            // of `Snapshot.blocks` guarantees that all in-block
+            // ancestors of any tx are visited before it.
+            let mut desc_set: FxHashSet<TxidPrefix> = FxHashSet::default();
+            desc_set.insert(*prefix);
+            for &i in &snapshot.blocks[seed_block.as_usize()] {
+                if descendants.len() >= MAX {
+                    break;
+                }
+                let Some(e) = entries.slot(i) else { continue };
+                if !e.depends.iter().any(|d| desc_set.contains(d)) {
+                    continue;
+                }
+                desc_set.insert(e.txid_prefix());
+                sum_fee += u64::from(e.fee);
+                sum_vsize += u64::from(e.vsize);
+                descendants.push(to_entry(e));
+            }
+        }
+
+        let best_descendant = descendants
+            .iter()
+            .max_by_key(|e| FeeRate::from((e.fee, e.weight)))
+            .cloned();
+
+        let package_rate = FeeRate::from((Sats::from(sum_fee), VSize::from(sum_vsize)));
+        let effective = seed.fee_rate().max(package_rate);
+
+        Some(CpfpInfo {
+            ancestors,
+            best_descendant,
+            descendants,
+            effective_fee_per_vsize: Some(effective),
+            fee: Some(seed.fee),
+            adjusted_vsize: Some(seed.vsize),
+        })
+    }
+}
+
+fn to_entry(e: &TxEntry) -> CpfpEntry {
+    CpfpEntry {
+        txid: e.txid.clone(),
+        weight: Weight::from(e.vsize),
+        fee: e.fee,
+    }
+}

crates/brk_mempool/src/lib.rs

@@ -21,13 +21,14 @@ use brk_types::{AddrBytes, MempoolInfo, TxOut, Txid, Vout};
 use parking_lot::RwLockReadGuard;
 use tracing::error;
 
+mod cpfp;
 pub(crate) mod steps;
 pub(crate) mod stores;
 #[cfg(test)]
 mod tests;
 
 use steps::{Applier, Fetcher, Preparer, Rebuilder, Resolver};
-pub use steps::{BlockStats, RecommendedFees, Snapshot, TxEntry, TxRemoval};
+pub use steps::{BlkIndex, BlockStats, RecommendedFees, Snapshot, TxEntry, TxRemoval};
 use stores::{AddrTracker, MempoolState};
 pub use stores::{EntryPool, TxGraveyard, TxStore, TxTombstone};
 

crates/brk_mempool/src/steps/mod.rs

@@ -9,5 +9,5 @@ mod resolver;
 pub use applier::Applier;
 pub use fetcher::Fetcher;
 pub use preparer::{Preparer, TxEntry, TxRemoval};
-pub use rebuilder::{BlockStats, Rebuilder, RecommendedFees, Snapshot};
+pub use rebuilder::{BlkIndex, BlockStats, Rebuilder, RecommendedFees, Snapshot};
 pub use resolver::Resolver;

crates/brk_mempool/src/steps/rebuilder/mod.rs

@@ -26,7 +26,7 @@ mod snapshot;
 mod verify;
 
 pub use brk_types::RecommendedFees;
-pub use snapshot::{BlockStats, Snapshot};
+pub use snapshot::{BlkIndex, BlockStats, Snapshot};
 
 const MIN_REBUILD_INTERVAL: Duration = Duration::from_secs(1);
 const NUM_BLOCKS: usize = 8;

crates/brk_mempool/src/steps/rebuilder/snapshot/blk_index.rs

@@ -0,0 +1,26 @@
+/// Projected-block index in a mempool snapshot. `u8` because the
+/// projection horizon is ~8 blocks at typical loads; `BlkIndex::MAX`
+/// is reserved as the "not in any projected block" sentinel used by
+/// `Snapshot::block_of` for txs below the mempool floor.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct BlkIndex(u8);
+
+impl BlkIndex {
+    /// Sentinel for "not in any projected block".
+    pub const MAX: BlkIndex = BlkIndex(u8::MAX);
+
+    pub fn is_not_in_projected(self) -> bool {
+        self == Self::MAX
+    }
+
+    pub fn as_usize(self) -> usize {
+        self.0 as usize
+    }
+}
+
+impl From<usize> for BlkIndex {
+    fn from(v: usize) -> Self {
+        debug_assert!(v < u8::MAX as usize, "BlkIndex overflow: {v}");
+        Self(v as u8)
+    }
+}

crates/brk_mempool/src/steps/rebuilder/snapshot/mod.rs

@@ -1,6 +1,8 @@
+mod blk_index;
 mod fees;
 mod stats;
 
+pub use blk_index::BlkIndex;
 pub use stats::BlockStats;
 
 use std::hash::{DefaultHasher, Hash, Hasher};
@@ -15,6 +17,10 @@ use fees::Fees;
 #[derive(Debug, Clone, Default)]
 pub struct Snapshot {
     pub blocks: Vec<Vec<TxIndex>>,
+    /// Reverse of `blocks`: indexed by `TxIndex.as_usize()`. Slots that
+    /// hold no entry, or hold an entry that didn't make any projected
+    /// block, store `BlkIndex::MAX`. Read via the `block_of` accessor.
+    block_of: Vec<BlkIndex>,
     pub block_stats: Vec<BlockStats>,
     pub fees: RecommendedFees,
     /// ETag-like cache key for the first projected block. A hash of
@@ -30,10 +36,12 @@ impl Snapshot {
         let block_stats = Self::compute_block_stats(&blocks, entries);
         let fees = Fees::compute(&block_stats, min_fee);
         let blocks = Self::flatten_blocks(blocks);
+        let block_of = Self::build_block_of(&blocks, entries.len());
         let next_block_hash = Self::hash_next_block(&blocks);
 
         Self {
             blocks,
+            block_of,
             block_stats,
             fees,
             next_block_hash,
@@ -60,6 +68,20 @@ impl Snapshot {
             .collect()
     }
 
+    /// One pass over `blocks` to invert the mapping. `BlkIndex::MAX`
+    /// stays as the sentinel for slots that aren't in any projected
+    /// block (empty slots and below-floor txs alike).
+    fn build_block_of(blocks: &[Vec<TxIndex>], entry_count: usize) -> Vec<BlkIndex> {
+        let mut block_of = vec![BlkIndex::MAX; entry_count];
+        for (b, txs) in blocks.iter().enumerate() {
+            let blk = BlkIndex::from(b);
+            for &idx in txs {
+                block_of[idx.as_usize()] = blk;
+            }
+        }
+        block_of
+    }
+
     fn hash_next_block(blocks: &[Vec<TxIndex>]) -> u64 {
         let Some(block) = blocks.first() else {
             return 0;
@@ -68,4 +90,13 @@ impl Snapshot {
         block.hash(&mut hasher);
         hasher.finish()
     }
+
+    /// Projected block that holds `idx`, or `None` if the tx is below
+    /// the mempool floor (or `idx` is out of range).
+    pub fn block_of(&self, idx: TxIndex) -> Option<BlkIndex> {
+        self.block_of
+            .get(idx.as_usize())
+            .copied()
+            .filter(|b| !b.is_not_in_projected())
+    }
 }

crates/brk_mempool/src/stores/entry_pool/mod.rs

@@ -1,6 +1,5 @@
 use brk_types::TxidPrefix;
 use rustc_hash::FxHashMap;
-use smallvec::SmallVec;
 
 mod tx_index;
 
@@ -40,19 +39,18 @@ impl EntryPool {
     }
 
     pub fn get(&self, prefix: &TxidPrefix) -> Option<&TxEntry> {
-        let idx = self.prefix_to_idx.get(prefix)?;
-        self.entries.get(idx.as_usize())?.as_ref()
+        self.slot(self.idx_of(prefix)?)
     }
 
-    /// Direct children of a transaction (txs whose `depends` includes
-    /// `prefix`). Linear scan over all entries.
-    pub fn children(&self, prefix: &TxidPrefix) -> SmallVec<[TxidPrefix; 2]> {
-        self.entries
-            .iter()
-            .flatten()
-            .filter(|e| e.depends.iter().any(|p| p == prefix))
-            .map(TxEntry::txid_prefix)
-            .collect()
+    /// Slot index for a prefix, or `None` if not in the pool.
+    pub fn idx_of(&self, prefix: &TxidPrefix) -> Option<TxIndex> {
+        self.prefix_to_idx.get(prefix).copied()
+    }
+
+    /// Direct slot read by index. `None` if the slot is empty or the
+    /// index is out of range.
+    pub fn slot(&self, idx: TxIndex) -> Option<&TxEntry> {
+        self.entries.get(idx.as_usize())?.as_ref()
     }
 
     pub fn remove(&mut self, prefix: &TxidPrefix) -> Option<TxEntry> {