Merge pull request #465 from tnull/2025-02-fix-mempool-access

bitcoind RPC: Make mempool syncing more efficient

Author: tnull

src/chain/bitcoind_rpc.rs

@@ -30,6 +30,8 @@ use std::sync::Arc;
 pub struct BitcoindRpcClient {
 	rpc_client: Arc<RpcClient>,
 	latest_mempool_timestamp: AtomicU64,
+	mempool_entries_cache: tokio::sync::Mutex<HashMap<Txid, MempoolEntry>>,
+	mempool_txs_cache: tokio::sync::Mutex<HashMap<Txid, (Transaction, u64)>>,
 }
 
 impl BitcoindRpcClient {
@@ -42,7 +44,9 @@ impl BitcoindRpcClient {
 
 		let latest_mempool_timestamp = AtomicU64::new(0);
 
-		Self { rpc_client, latest_mempool_timestamp }
+		let mempool_entries_cache = tokio::sync::Mutex::new(HashMap::new());
+		let mempool_txs_cache = tokio::sync::Mutex::new(HashMap::new());
+		Self { rpc_client, latest_mempool_timestamp, mempool_entries_cache, mempool_txs_cache }
 	}
 
 	pub(crate) fn rpc_client(&self) -> Arc<RpcClient> {
@@ -122,23 +126,68 @@ impl BitcoindRpcClient {
 			.map(|resp| resp.0)
 	}
 
-	pub(crate) async fn get_mempool_entry(&self, txid: Txid) -> std::io::Result<MempoolEntry> {
+	pub(crate) async fn get_mempool_entry(
+		&self, txid: Txid,
+	) -> std::io::Result<Option<MempoolEntry>> {
 		let txid_hex = bitcoin::consensus::encode::serialize_hex(&txid);
 		let txid_json = serde_json::json!(txid_hex);
-		self.rpc_client
+		match self
+			.rpc_client
 			.call_method::<GetMempoolEntryResponse>("getmempoolentry", &[txid_json])
 			.await
-			.map(|resp| MempoolEntry { txid, height: resp.height, time: resp.time })
+		{
+			Ok(resp) => Ok(Some(MempoolEntry { txid, height: resp.height, time: resp.time })),
+			Err(e) => match e.into_inner() {
+				Some(inner) => {
+					let rpc_error_res: Result<Box<RpcError>, _> = inner.downcast();
+
+					match rpc_error_res {
+						Ok(rpc_error) => {
+							// Check if it's the 'not found' error code.
+							if rpc_error.code == -5 {
+								Ok(None)
+							} else {
+								Err(std::io::Error::new(std::io::ErrorKind::Other, rpc_error))
+							}
+						},
+						Err(_) => Err(std::io::Error::new(
+							std::io::ErrorKind::Other,
+							"Failed to process getmempoolentry response",
+						)),
+					}
+				},
+				None => Err(std::io::Error::new(
+					std::io::ErrorKind::Other,
+					"Failed to process getmempoolentry response",
+				)),
+			},
+		}
 	}
 
-	pub(crate) async fn get_mempool_entries(&self) -> std::io::Result<Vec<MempoolEntry>> {
+	pub(crate) async fn update_mempool_entries_cache(&self) -> std::io::Result<()> {
 		let mempool_txids = self.get_raw_mempool().await?;
-		let mut mempool_entries = Vec::with_capacity(mempool_txids.len());
+
+		let mut mempool_entries_cache = self.mempool_entries_cache.lock().await;
+		mempool_entries_cache.retain(|txid, _| mempool_txids.contains(txid));
+
+		if let Some(difference) = mempool_txids.len().checked_sub(mempool_entries_cache.capacity())
+		{
+			mempool_entries_cache.reserve(difference)
+		}
+
 		for txid in mempool_txids {
-			let entry = self.get_mempool_entry(txid).await?;
-			mempool_entries.push(entry);
+			if mempool_entries_cache.contains_key(&txid) {
+				continue;
+			}
+
+			if let Some(entry) = self.get_mempool_entry(txid).await? {
+				mempool_entries_cache.insert(txid, entry.clone());
+			}
 		}
-		Ok(mempool_entries)
+
+		mempool_entries_cache.shrink_to_fit();
+
+		Ok(())
 	}
 
 	/// Get mempool transactions, alongside their first-seen unix timestamps.
@@ -152,10 +201,20 @@ impl BitcoindRpcClient {
 		let prev_mempool_time = self.latest_mempool_timestamp.load(Ordering::Relaxed);
 		let mut latest_time = prev_mempool_time;
 
-		let mempool_entries = self.get_mempool_entries().await?;
-		let mut txs_to_emit = Vec::new();
+		self.update_mempool_entries_cache().await?;
+
+		let mempool_entries_cache = self.mempool_entries_cache.lock().await;
+		let mut mempool_txs_cache = self.mempool_txs_cache.lock().await;
+		mempool_txs_cache.retain(|txid, _| mempool_entries_cache.contains_key(txid));
 
-		for entry in mempool_entries {
+		if let Some(difference) =
+			mempool_entries_cache.len().checked_sub(mempool_txs_cache.capacity())
+		{
+			mempool_txs_cache.reserve(difference)
+		}
+
+		let mut txs_to_emit = Vec::with_capacity(mempool_entries_cache.len());
+		for (txid, entry) in mempool_entries_cache.iter() {
 			if entry.time > latest_time {
 				latest_time = entry.time;
 			}
@@ -171,8 +230,14 @@ impl BitcoindRpcClient {
 				continue;
 			}
 
+			if let Some((cached_tx, cached_time)) = mempool_txs_cache.get(txid) {
+				txs_to_emit.push((cached_tx.clone(), *cached_time));
+				continue;
+			}
+
 			match self.get_raw_transaction(&entry.txid).await {
 				Ok(Some(tx)) => {
+					mempool_txs_cache.insert(entry.txid, (tx.clone(), entry.time));
 					txs_to_emit.push((tx, entry.time));
 				},
 				Ok(None) => {

src/chain/mod.rs

@@ -317,8 +317,14 @@ impl ChainSource {
 					));
 				}
 
+				log_info!(
+					logger,
+					"Starting initial synchronization of chain listeners. This might take a while..",
+				);
+
 				loop {
 					let mut locked_header_cache = header_cache.lock().await;
+					let now = SystemTime::now();
 					match synchronize_listeners(
 						bitcoind_rpc_client.as_ref(),
 						config.network,
@@ -329,6 +335,11 @@ impl ChainSource {
 					{
 						Ok(chain_tip) => {
 							{
+								log_info!(
+									logger,
+									"Finished synchronizing listeners in {}ms",
+									now.elapsed().unwrap().as_millis()
+								);
 								*latest_chain_tip.write().unwrap() = Some(chain_tip);
 								let unix_time_secs_opt = SystemTime::now()
 									.duration_since(UNIX_EPOCH)
@@ -374,6 +385,8 @@ impl ChainSource {
 				fee_rate_update_interval
 					.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
+				log_info!(logger, "Starting continuous polling for chain updates.");
+
 				// Start the polling loop.
 				loop {
 					tokio::select! {
@@ -692,13 +705,15 @@ impl ChainSource {
 					&mut *locked_header_cache,
 					&chain_listener,
 				);
-				let mut chain_polling_interval =
-					tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS));
-				chain_polling_interval
-					.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
+				let now = SystemTime::now();
 				match spv_client.poll_best_tip().await {
 					Ok((ChainTip::Better(tip), true)) => {
+						log_trace!(
+							logger,
+							"Finished polling best tip in {}ms",
+							now.elapsed().unwrap().as_millis()
+						);
 						*latest_chain_tip.write().unwrap() = Some(tip);
 					},
 					Ok(_) => {},
@@ -711,11 +726,19 @@ impl ChainSource {
 				}
 
 				let cur_height = channel_manager.current_best_block().height;
+
+				let now = SystemTime::now();
 				match bitcoind_rpc_client
 					.get_mempool_transactions_and_timestamp_at_height(cur_height)
 					.await
 				{
 					Ok(unconfirmed_txs) => {
+						log_trace!(
+							logger,
+							"Finished polling mempool of size {} in {}ms",
+							unconfirmed_txs.len(),
+							now.elapsed().unwrap().as_millis()
+						);
 						let _ = onchain_wallet.apply_unconfirmed_txs(unconfirmed_txs);
 					},
 					Err(e) => {