feat(index,node): wire bitcoin_rs_index::Indexer into apply_block via IndexerLike trait object

Op: extend

Author: metaphorics

crates/index/src/index.rs

@@ -199,3 +199,18 @@ impl Visitor for IndexBlockVisitor<'_> {
 fn is_null_prevout(prevout: &bsl::OutPoint<'_>) -> bool {
     prevout.vout() == u32::MAX && prevout.txid().iter().all(|byte| *byte == 0)
 }
+
+/// Storage-agnostic block-ingest interface.
+///
+/// Use this trait when consumers must hold the indexer behind a trait
+/// object (e.g. when the storage backend is selected at runtime).
+pub trait IndexerLike: Send + Sync {
+    /// Walks `block` once and writes index rows. See `Indexer::ingest_block`.
+    fn ingest_block(&mut self, block: &[u8], height: u32) -> Result<IndexRowCounts, IndexError>;
+}
+
+impl<S: KvStore + Send + Sync + 'static> IndexerLike for Indexer<S> {
+    fn ingest_block(&mut self, block: &[u8], height: u32) -> Result<IndexRowCounts, IndexError> {
+        Self::ingest_block(self, block, height)
+    }
+}

crates/index/src/lib.rs

@@ -10,7 +10,7 @@ pub mod status;
 /// Stable electrs-shaped row types.
 pub mod types;
 
-pub use index::{IndexError, IndexRowCounts, Indexer};
+pub use index::{IndexError, IndexRowCounts, Indexer, IndexerLike};
 pub use mempool::{MempoolRowCounts, MempoolRowWriter};
 pub use status::{HistoryEntry, HistoryHeight, StatusHash, compute_status_hash};
 pub use types::{

crates/node/src/apply.rs

@@ -37,6 +37,8 @@ pub struct ApplyHandles {
     pub utxo: Arc<UtxoSet>,
     /// Shared coinstats listener.
     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 mempool.
     pub mempool: Arc<RwLock<Mempool>>,
     /// Shared block records exposed to RPC handlers.
@@ -215,6 +217,31 @@ pub fn apply_block(
     let coin_stats_dur = coin_stats_started.elapsed();
     metrics::histogram!("node.apply_block.coin_stats_finish_seconds")
         .record(coin_stats_dur.as_secs_f64());
+    let tx_index_ingest_started = quanta::Instant::now();
+    let block_bytes = bitcoin::consensus::encode::serialize(block);
+    let tx_index_ingest_result = handles.tx_index.lock().ingest_block(&block_bytes, height);
+    match tx_index_ingest_result {
+        Ok(counts) => {
+            tracing::debug!(
+                height,
+                txids = counts.txids,
+                funding = counts.funding,
+                spending = counts.spending,
+                headers = counts.headers,
+                "tx_index ingested block"
+            );
+        }
+        Err(error) => {
+            tracing::warn!(
+                height,
+                %error,
+                "tx_index failed to ingest block; best-effort path continues"
+            );
+        }
+    }
+    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 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);
@@ -234,6 +261,7 @@ pub fn apply_block(
         block_tree_insert_us = block_tree_insert_dur.as_micros(),
         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(),
         coin_stats_us = coin_stats_dur.as_micros(),
         total_us = total_dur.as_micros(),
         "apply_block: profile"

crates/node/src/state.rs

@@ -23,6 +23,8 @@ use parking_lot::{Mutex, RwLock};
 
 use crate::Config;
 
+type TxIndexHandle = Arc<Mutex<Box<dyn bitcoin_rs_index::IndexerLike>>>;
+
 /// Errors produced when applying a block to the node state.
 #[derive(Debug, thiserror::Error)]
 pub enum ApplyError {
@@ -165,13 +167,48 @@ impl fmt::Display for CompiledStorageFeatures {
     }
 }
 
+fn open_tx_index(config: &Config) -> Result<TxIndexHandle> {
+    let txindex_dir = config.data_dir.join("txindex");
+    std::fs::create_dir_all(&txindex_dir)
+        .with_context(|| format!("create txindex_dir {}", txindex_dir.display()))?;
+    let tx_index: Box<dyn bitcoin_rs_index::IndexerLike> = match config.storage_backend.as_str() {
+        #[cfg(feature = "rocksdb")]
+        "rocksdb" => {
+            let store =
+                bitcoin_rs_storage::RocksDbStore::open(&txindex_dir).map_err(anyhow::Error::new)?;
+            Box::new(bitcoin_rs_index::Indexer::new(Arc::new(store)))
+        }
+        #[cfg(feature = "fjall")]
+        "fjall" => {
+            let store =
+                bitcoin_rs_storage::FjallStore::open(&txindex_dir).map_err(anyhow::Error::new)?;
+            Box::new(bitcoin_rs_index::Indexer::new(Arc::new(store)))
+        }
+        #[cfg(feature = "redb")]
+        "redb" => {
+            let store =
+                bitcoin_rs_storage::RedbStore::open(&txindex_dir).map_err(anyhow::Error::new)?;
+            Box::new(bitcoin_rs_index::Indexer::new(Arc::new(store)))
+        }
+        #[cfg(feature = "mdbx")]
+        "mdbx" => {
+            let store =
+                bitcoin_rs_storage::MdbxStore::open(&txindex_dir).map_err(anyhow::Error::new)?;
+            Box::new(bitcoin_rs_index::Indexer::new(Arc::new(store)))
+        }
+        other => bail!("unsupported storage backend for txindex: {other}"),
+    };
+    Ok(Arc::new(Mutex::new(tx_index)))
+}
+
 /// Aggregate handle to a running node.
 pub struct NodeState {
     config: Config,
     data_dir: PathBuf,
     storage: NodeStorage,
     utxo: Arc<UtxoSet>,
     coin_stats: Arc<bitcoin_rs_coinstats::CoinStatsListener>,
+    tx_index: TxIndexHandle,
     mempool: Arc<RwLock<Mempool>>,
     chain_tip: Arc<ArcSwapOption<TipSnapshot>>,
     applied_tip: Arc<ArcSwapOption<TipSnapshot>>,
@@ -203,6 +240,7 @@ impl NodeState {
         std::fs::create_dir_all(&config.data_dir)
             .with_context(|| format!("create data_dir {}", config.data_dir.display()))?;
         let storage = NodeStorage::open(&config)?;
+        let tx_index = open_tx_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(),
@@ -234,6 +272,7 @@ impl NodeState {
                 block_tree: Arc::clone(&block_tree),
                 utxo: Arc::clone(&utxo),
                 coin_stats: Arc::clone(&coin_stats),
+                tx_index: Arc::clone(&tx_index),
                 mempool: Arc::clone(&mempool),
                 blocks: Arc::clone(&blocks),
                 transactions: Arc::clone(&transactions),
@@ -255,6 +294,7 @@ impl NodeState {
             storage,
             utxo,
             coin_stats,
+            tx_index,
             mempool,
             chain_tip,
             applied_tip,
@@ -304,6 +344,12 @@ impl NodeState {
         Arc::clone(&self.coin_stats)
     }
 
+    /// Returns the shared block indexer handle.
+    #[must_use]
+    pub fn tx_index(&self) -> Arc<Mutex<Box<dyn bitcoin_rs_index::IndexerLike>>> {
+        Arc::clone(&self.tx_index)
+    }
+
     /// Returns the shared mempool handle.
     #[must_use]
     pub fn mempool(&self) -> Arc<RwLock<Mempool>> {
@@ -451,6 +497,7 @@ impl NodeState {
             block_tree: Arc::clone(&self.block_tree),
             utxo: Arc::clone(&self.utxo),
             coin_stats: Arc::clone(&self.coin_stats),
+            tx_index: Arc::clone(&self.tx_index),
             mempool: Arc::clone(&self.mempool),
             blocks: Arc::clone(&self.blocks),
             transactions: Arc::clone(&self.transactions),
@@ -538,6 +585,19 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn open_constructs_tx_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.tx_index();
+        let b = state.tx_index();
+        assert!(Arc::ptr_eq(&a, &b), "tx_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_index::{IndexError, IndexRowCounts, IndexerLike};
     use bitcoin_rs_mempool::{Mempool, MempoolLimits};
     use bitcoin_rs_p2p::PeerInfo;
     use bitcoin_rs_utxo::UtxoSet;
@@ -480,12 +481,30 @@ mod tests {
             coin_stats: Arc::new(bitcoin_rs_coinstats::CoinStatsListener::new(
                 bitcoin_rs_coinstats::CoinStats::default(),
             )),
+            tx_index: noop_tx_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())),
         }
     }
 
+    struct NoopIndexer;
+
+    impl IndexerLike for NoopIndexer {
+        fn ingest_block(
+            &mut self,
+            _block: &[u8],
+            _height: u32,
+        ) -> Result<IndexRowCounts, IndexError> {
+            Ok(IndexRowCounts::default())
+        }
+    }
+
+    fn noop_tx_index() -> Arc<Mutex<Box<dyn IndexerLike>>> {
+        let indexer: Box<dyn IndexerLike> = Box::new(NoopIndexer);
+        Arc::new(Mutex::new(indexer))
+    }
+
     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_index::{IndexError, IndexRowCounts, IndexerLike};
 use bitcoin_rs_mempool::{Mempool, MempoolLimits};
 use bitcoin_rs_node::{BlockSync, Network, apply::ApplyHandles};
 use bitcoin_rs_p2p::{Message, PeerInfo};
@@ -160,12 +161,26 @@ fn apply_handles(
         coin_stats: Arc::new(bitcoin_rs_coinstats::CoinStatsListener::new(
             bitcoin_rs_coinstats::CoinStats::default(),
         )),
+        tx_index: noop_tx_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())),
     }
 }
 
+struct NoopIndexer;
+
+impl IndexerLike for NoopIndexer {
+    fn ingest_block(&mut self, _block: &[u8], _height: u32) -> Result<IndexRowCounts, IndexError> {
+        Ok(IndexRowCounts::default())
+    }
+}
+
+fn noop_tx_index() -> Arc<Mutex<Box<dyn IndexerLike>>> {
+    let indexer: Box<dyn IndexerLike> = Box::new(NoopIndexer);
+    Arc::new(Mutex::new(indexer))
+}
+
 fn regtest_genesis_block() -> Result<bitcoin::Block, Box<dyn std::error::Error>> {
     use bitcoin::consensus::Decodable as _;