feat(filters,node): wire FilterIndex into apply_block via FilterIndexLike trait object

Op: extend

Author: metaphorics

crates/filters/src/filter_index.rs

@@ -87,3 +87,38 @@ impl<S: KvStore> FilterIndex<S> {
         Ok(Some(Hash256::from_le_bytes(&header)))
     }
 }
+
+/// Storage-agnostic compact-filter ingest interface.
+pub trait FilterIndexLike: Send + Sync {
+    /// Stores a block filter and returns its chained BIP157 filter header.
+    fn put_filter(
+        &self,
+        block_hash: bitcoin_rs_primitives::Hash256,
+        prev_header: bitcoin_rs_primitives::Hash256,
+        filter_bytes: &[u8],
+    ) -> Result<bitcoin_rs_primitives::Hash256, FilterIndexError>;
+
+    /// Loads the BIP157 filter header for a block, if indexed.
+    fn filter_header(
+        &self,
+        block_hash: bitcoin_rs_primitives::Hash256,
+    ) -> Result<Option<bitcoin_rs_primitives::Hash256>, FilterIndexError>;
+}
+
+impl<S: KvStore + Send + Sync + 'static> FilterIndexLike for FilterIndex<S> {
+    fn put_filter(
+        &self,
+        block_hash: bitcoin_rs_primitives::Hash256,
+        prev_header: bitcoin_rs_primitives::Hash256,
+        filter_bytes: &[u8],
+    ) -> Result<bitcoin_rs_primitives::Hash256, FilterIndexError> {
+        Self::put_filter(self, block_hash, prev_header, filter_bytes)
+    }
+
+    fn filter_header(
+        &self,
+        block_hash: bitcoin_rs_primitives::Hash256,
+    ) -> Result<Option<bitcoin_rs_primitives::Hash256>, FilterIndexError> {
+        Self::filter_header(self, block_hash)
+    }
+}

crates/filters/src/lib.rs

@@ -8,4 +8,4 @@ pub mod filter_index;
 /// BIP158 Golomb-coded set codec.
 pub mod gcs;
 
-pub use filter_index::{FilterIndex, FilterIndexError};
+pub use filter_index::{FilterIndex, FilterIndexError, FilterIndexLike};

crates/node/src/apply.rs

@@ -39,6 +39,8 @@ pub struct ApplyHandles {
     pub coin_stats: Arc<bitcoin_rs_coinstats::CoinStatsListener>,
     /// Shared best-effort confirmed transaction indexer.
     pub tx_index: Arc<parking_lot::Mutex<Box<dyn bitcoin_rs_index::IndexerLike>>>,
+    /// Shared best-effort compact-filter indexer.
+    pub filter_index: Arc<Box<dyn bitcoin_rs_filters::FilterIndexLike>>,
     /// Shared mempool.
     pub mempool: Arc<RwLock<Mempool>>,
     /// Shared block records exposed to RPC handlers.
@@ -159,6 +161,13 @@ pub fn apply_block(
     metrics::histogram!("node.apply_block.bip68_seconds").record(bip68_dur.as_secs_f64());
     bip68_result?;
 
+    let filter_bytes = compute_basic_filter(block, handles).unwrap_or_else(|| {
+        tracing::trace!(
+            "BIP158 filter generation unavailable; storing empty filter as placeholder"
+        );
+        Vec::new()
+    });
+
     let changes = build_utxo_changes(block, height)?;
     let utxo_commit_started = quanta::Instant::now();
     let utxo_commit_result = handles.utxo.commit_block(&changes, &block_hash);
@@ -242,6 +251,30 @@ pub fn apply_block(
     let tx_index_ingest_dur = tx_index_ingest_started.elapsed();
     metrics::histogram!("node.apply_block.tx_index_ingest_seconds")
         .record(tx_index_ingest_dur.as_secs_f64());
+    let filter_started = quanta::Instant::now();
+    let prev_filter_header = handles
+        .applied_tip
+        .load_full()
+        .and_then(|tip| handles.filter_index.filter_header(tip.hash).ok().flatten())
+        .unwrap_or_default();
+    match handles
+        .filter_index
+        .put_filter(block_hash, prev_filter_header, &filter_bytes)
+    {
+        Ok(filter_header) => {
+            tracing::debug!(
+                height,
+                %filter_header,
+                bytes = filter_bytes.len(),
+                "filter_index stored block filter"
+            );
+        }
+        Err(error) => {
+            tracing::warn!(height, %error, "filter_index failed to store block filter");
+        }
+    }
+    let filter_dur = filter_started.elapsed();
+    metrics::histogram!("node.apply_block.filter_index_seconds").record(filter_dur.as_secs_f64());
     let total_dur = total_started.elapsed();
     metrics::histogram!("node.apply_block.total_seconds").record(total_dur.as_secs_f64());
     metrics::counter!("node.apply_block.txs_applied").increment(tx_count_delta);
@@ -262,6 +295,7 @@ pub fn apply_block(
         mempool_evict_us = mempool_evict_dur.as_micros(),
         tx_index_us = tx_index_dur.as_micros(),
         tx_index_ingest_us = tx_index_ingest_dur.as_micros(),
+        filter_index_us = filter_dur.as_micros(),
         coin_stats_us = coin_stats_dur.as_micros(),
         total_us = total_dur.as_micros(),
         "apply_block: profile"
@@ -281,6 +315,24 @@ fn insert_active_header(
     Ok(())
 }
 
+fn compute_basic_filter(block: &bitcoin::Block, handles: &ApplyHandles) -> Option<Vec<u8>> {
+    use bitcoin::hashes::Hash as _;
+
+    let filter = bitcoin::bip158::BlockFilter::new_script_filter(block, |outpoint| {
+        let prev_outpoint = OutPoint::new(
+            bitcoin_rs_primitives::Hash256::from_le_bytes(outpoint.txid.as_byte_array()),
+            outpoint.vout,
+        );
+        handles
+            .utxo
+            .get(&prev_outpoint)
+            .map(|txout| txout.script_pubkey)
+            .ok_or(bitcoin::bip158::Error::UtxoMissing(*outpoint))
+    })
+    .ok()?;
+    Some(filter.content)
+}
+
 fn verify_block_transactions(
     handles: &ApplyHandles,
     block: &bitcoin::Block,

crates/node/src/state.rs

@@ -24,6 +24,7 @@ use parking_lot::{Mutex, RwLock};
 use crate::Config;
 
 type TxIndexHandle = Arc<Mutex<Box<dyn bitcoin_rs_index::IndexerLike>>>;
+type FilterIndexHandle = Arc<Box<dyn bitcoin_rs_filters::FilterIndexLike>>;
 
 /// Errors produced when applying a block to the node state.
 #[derive(Debug, thiserror::Error)]
@@ -201,6 +202,33 @@ fn open_tx_index(config: &Config) -> Result<TxIndexHandle> {
     Ok(Arc::new(Mutex::new(tx_index)))
 }
 
+fn open_filter_index(config: &Config) -> Result<FilterIndexHandle> {
+    let filters_dir = config.data_dir.join("filters");
+    std::fs::create_dir_all(&filters_dir)
+        .with_context(|| format!("create filters_dir {}", filters_dir.display()))?;
+    let filter_index: Box<dyn bitcoin_rs_filters::FilterIndexLike> =
+        match config.storage_backend.as_str() {
+            #[cfg(feature = "rocksdb")]
+            "rocksdb" => Box::new(bitcoin_rs_filters::FilterIndex::new(
+                bitcoin_rs_storage::RocksDbStore::open(&filters_dir).map_err(anyhow::Error::new)?,
+            )),
+            #[cfg(feature = "fjall")]
+            "fjall" => Box::new(bitcoin_rs_filters::FilterIndex::new(
+                bitcoin_rs_storage::FjallStore::open(&filters_dir).map_err(anyhow::Error::new)?,
+            )),
+            #[cfg(feature = "redb")]
+            "redb" => Box::new(bitcoin_rs_filters::FilterIndex::new(
+                bitcoin_rs_storage::RedbStore::open(&filters_dir).map_err(anyhow::Error::new)?,
+            )),
+            #[cfg(feature = "mdbx")]
+            "mdbx" => Box::new(bitcoin_rs_filters::FilterIndex::new(
+                bitcoin_rs_storage::MdbxStore::open(&filters_dir).map_err(anyhow::Error::new)?,
+            )),
+            other => bail!("unsupported storage backend for filter index: {other}"),
+        };
+    Ok(Arc::new(filter_index))
+}
+
 /// Aggregate handle to a running node.
 pub struct NodeState {
     config: Config,
@@ -209,6 +237,7 @@ pub struct NodeState {
     utxo: Arc<UtxoSet>,
     coin_stats: Arc<bitcoin_rs_coinstats::CoinStatsListener>,
     tx_index: TxIndexHandle,
+    filter_index: FilterIndexHandle,
     mempool: Arc<RwLock<Mempool>>,
     chain_tip: Arc<ArcSwapOption<TipSnapshot>>,
     applied_tip: Arc<ArcSwapOption<TipSnapshot>>,
@@ -241,6 +270,7 @@ impl NodeState {
             .with_context(|| format!("create data_dir {}", config.data_dir.display()))?;
         let storage = NodeStorage::open(&config)?;
         let tx_index = open_tx_index(&config)?;
+        let filter_index = open_filter_index(&config)?;
         let mut utxo_set = bitcoin_rs_utxo::UtxoSet::new();
         let coin_stats_listener = bitcoin_rs_coinstats::CoinStatsListener::new(
             bitcoin_rs_coinstats::CoinStats::default(),
@@ -273,6 +303,7 @@ impl NodeState {
                 utxo: Arc::clone(&utxo),
                 coin_stats: Arc::clone(&coin_stats),
                 tx_index: Arc::clone(&tx_index),
+                filter_index: Arc::clone(&filter_index),
                 mempool: Arc::clone(&mempool),
                 blocks: Arc::clone(&blocks),
                 transactions: Arc::clone(&transactions),
@@ -295,6 +326,7 @@ impl NodeState {
             utxo,
             coin_stats,
             tx_index,
+            filter_index,
             mempool,
             chain_tip,
             applied_tip,
@@ -350,6 +382,12 @@ impl NodeState {
         Arc::clone(&self.tx_index)
     }
 
+    /// Returns the shared compact-filter index handle.
+    #[must_use]
+    pub fn filter_index(&self) -> FilterIndexHandle {
+        Arc::clone(&self.filter_index)
+    }
+
     /// Returns the shared mempool handle.
     #[must_use]
     pub fn mempool(&self) -> Arc<RwLock<Mempool>> {
@@ -498,6 +536,7 @@ impl NodeState {
             utxo: Arc::clone(&self.utxo),
             coin_stats: Arc::clone(&self.coin_stats),
             tx_index: Arc::clone(&self.tx_index),
+            filter_index: Arc::clone(&self.filter_index),
             mempool: Arc::clone(&self.mempool),
             blocks: Arc::clone(&self.blocks),
             transactions: Arc::clone(&self.transactions),
@@ -598,6 +637,22 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn open_constructs_filter_index() -> anyhow::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let mut config = crate::Config::default_for_network(crate::Network::Regtest);
+        config.data_dir = dir.path().join("node");
+        config.p2p_listen.clear();
+        let state = NodeState::open(config)?;
+        let a = state.filter_index();
+        let b = state.filter_index();
+        assert!(
+            Arc::ptr_eq(&a, &b),
+            "filter_index handle stable across calls"
+        );
+        Ok(())
+    }
+
     #[test]
     fn open_constructs_block_sync_orchestrator() -> anyhow::Result<()> {
         let dir = tempfile::tempdir()?;

crates/node/src/sync.rs

@@ -305,6 +305,7 @@ mod tests {
         pow::CompactTarget,
     };
     use bitcoin_rs_chain::{BlockTree, ChainWork, NodeStatus, TipSnapshot};
+    use bitcoin_rs_filters::{FilterIndexError, FilterIndexLike};
     use bitcoin_rs_index::{IndexError, IndexRowCounts, IndexerLike};
     use bitcoin_rs_mempool::{Mempool, MempoolLimits};
     use bitcoin_rs_p2p::PeerInfo;
@@ -482,6 +483,7 @@ mod tests {
                 bitcoin_rs_coinstats::CoinStats::default(),
             )),
             tx_index: noop_tx_index(),
+            filter_index: noop_filter_index(),
             mempool: Arc::new(RwLock::new(Mempool::new(MempoolLimits::default()))),
             blocks: Arc::new(RwLock::new(Vec::new())),
             transactions: Arc::new(RwLock::new(HashMap::<Txid, Transaction>::new())),
@@ -505,6 +507,31 @@ mod tests {
         Arc::new(Mutex::new(indexer))
     }
 
+    struct NoopFilterIndex;
+
+    impl FilterIndexLike for NoopFilterIndex {
+        fn put_filter(
+            &self,
+            _block_hash: bitcoin_rs_primitives::Hash256,
+            _prev_header: bitcoin_rs_primitives::Hash256,
+            _filter_bytes: &[u8],
+        ) -> Result<bitcoin_rs_primitives::Hash256, FilterIndexError> {
+            Ok(bitcoin_rs_primitives::Hash256::default())
+        }
+
+        fn filter_header(
+            &self,
+            _block_hash: bitcoin_rs_primitives::Hash256,
+        ) -> Result<Option<bitcoin_rs_primitives::Hash256>, FilterIndexError> {
+            Ok(None)
+        }
+    }
+
+    fn noop_filter_index() -> Arc<Box<dyn FilterIndexLike>> {
+        let filter_index: Box<dyn FilterIndexLike> = Box::new(NoopFilterIndex);
+        Arc::new(filter_index)
+    }
+
     fn test_header(prev_blockhash: BlockHash, height: u32) -> BlockHeader {
         let mut merkle = [0_u8; 32];
         merkle[..4].copy_from_slice(&height.to_le_bytes());

crates/node/tests/sync_smoke.rs

@@ -7,6 +7,7 @@ use bitcoin::hashes::Hash as _;
 use bitcoin::p2p::message::NetworkMessage;
 use bitcoin::{BlockHash, Transaction, Txid};
 use bitcoin_rs_chain::{BlockTree, TipSnapshot};
+use bitcoin_rs_filters::{FilterIndexError, FilterIndexLike};
 use bitcoin_rs_index::{IndexError, IndexRowCounts, IndexerLike};
 use bitcoin_rs_mempool::{Mempool, MempoolLimits};
 use bitcoin_rs_node::{BlockSync, Network, apply::ApplyHandles};
@@ -162,6 +163,7 @@ fn apply_handles(
             bitcoin_rs_coinstats::CoinStats::default(),
         )),
         tx_index: noop_tx_index(),
+        filter_index: noop_filter_index(),
         mempool: Arc::new(RwLock::new(Mempool::new(MempoolLimits::default()))),
         blocks: Arc::new(RwLock::new(Vec::new())),
         transactions: Arc::new(RwLock::new(HashMap::<Txid, Transaction>::new())),
@@ -181,6 +183,31 @@ fn noop_tx_index() -> Arc<Mutex<Box<dyn IndexerLike>>> {
     Arc::new(Mutex::new(indexer))
 }
 
+struct NoopFilterIndex;
+
+impl FilterIndexLike for NoopFilterIndex {
+    fn put_filter(
+        &self,
+        _block_hash: bitcoin_rs_primitives::Hash256,
+        _prev_header: bitcoin_rs_primitives::Hash256,
+        _filter_bytes: &[u8],
+    ) -> Result<bitcoin_rs_primitives::Hash256, FilterIndexError> {
+        Ok(bitcoin_rs_primitives::Hash256::default())
+    }
+
+    fn filter_header(
+        &self,
+        _block_hash: bitcoin_rs_primitives::Hash256,
+    ) -> Result<Option<bitcoin_rs_primitives::Hash256>, FilterIndexError> {
+        Ok(None)
+    }
+}
+
+fn noop_filter_index() -> Arc<Box<dyn FilterIndexLike>> {
+    let filter_index: Box<dyn FilterIndexLike> = Box::new(NoopFilterIndex);
+    Arc::new(filter_index)
+}
+
 fn regtest_genesis_block() -> Result<bitcoin::Block, Box<dyn std::error::Error>> {
     use bitcoin::consensus::Decodable as _;