Add initial polling logic

We first initialize by synchronizing all `Listen` implementations, and
then head into a loop continuously polling our RPC `BlockSource`.

We also implement a `BoundedHeaderCache` to limit in-memory footprint.

Author: tnull

src/chain/bitcoind_rpc.rs

@@ -5,14 +5,22 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use crate::types::{ChainMonitor, ChannelManager, Sweeper, Wallet};
+
+use lightning::chain::Listen;
+
 use lightning_block_sync::http::HttpEndpoint;
+use lightning_block_sync::poll::ValidatedBlockHeader;
 use lightning_block_sync::rpc::RpcClient;
-use lightning_block_sync::{AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource};
+use lightning_block_sync::{
+	AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource, Cache,
+};
 
 use bitcoin::BlockHash;
 
 use base64::prelude::{Engine, BASE64_STANDARD};
 
+use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 
 pub struct BitcoindRpcClient {
@@ -51,3 +59,75 @@ impl BlockSource for BitcoindRpcClient {
 		Box::pin(async move { self.rpc_client.get_best_block().await })
 	}
 }
+
+const MAX_HEADER_CACHE_ENTRIES: usize = 100;
+
+pub(super) struct BoundedHeaderCache {
+	header_map: HashMap<BlockHash, ValidatedBlockHeader>,
+	recently_seen: VecDeque<BlockHash>,
+}
+
+impl BoundedHeaderCache {
+	pub(super) fn new() -> Self {
+		let header_map = HashMap::new();
+		let recently_seen = VecDeque::new();
+		Self { header_map, recently_seen }
+	}
+}
+
+impl Cache for BoundedHeaderCache {
+	fn look_up(&self, block_hash: &BlockHash) -> Option<&ValidatedBlockHeader> {
+		self.header_map.get(block_hash)
+	}
+
+	fn block_connected(&mut self, block_hash: BlockHash, block_header: ValidatedBlockHeader) {
+		self.recently_seen.push_back(block_hash);
+		self.header_map.insert(block_hash, block_header);
+
+		if self.header_map.len() >= MAX_HEADER_CACHE_ENTRIES {
+			// Keep dropping old entries until we've actually removed a header entry.
+			while let Some(oldest_entry) = self.recently_seen.pop_front() {
+				if self.header_map.remove(&oldest_entry).is_some() {
+					break;
+				}
+			}
+		}
+	}
+
+	fn block_disconnected(&mut self, block_hash: &BlockHash) -> Option<ValidatedBlockHeader> {
+		self.recently_seen.retain(|e| e != block_hash);
+		self.header_map.remove(block_hash)
+	}
+}
+
+pub(crate) struct ChainListener {
+	pub(crate) onchain_wallet: Arc<Wallet>,
+	pub(crate) channel_manager: Arc<ChannelManager>,
+	pub(crate) chain_monitor: Arc<ChainMonitor>,
+	pub(crate) output_sweeper: Arc<Sweeper>,
+}
+
+impl Listen for ChainListener {
+	fn filtered_block_connected(
+		&self, header: &bitcoin::block::Header,
+		txdata: &lightning::chain::transaction::TransactionData, height: u32,
+	) {
+		self.onchain_wallet.filtered_block_connected(header, txdata, height);
+		self.channel_manager.filtered_block_connected(header, txdata, height);
+		self.chain_monitor.filtered_block_connected(header, txdata, height);
+		self.output_sweeper.filtered_block_connected(header, txdata, height);
+	}
+	fn block_connected(&self, block: &bitcoin::Block, height: u32) {
+		self.onchain_wallet.block_connected(block, height);
+		self.channel_manager.block_connected(block, height);
+		self.chain_monitor.block_connected(block, height);
+		self.output_sweeper.block_connected(block, height);
+	}
+
+	fn block_disconnected(&self, header: &bitcoin::block::Header, height: u32) {
+		self.onchain_wallet.block_disconnected(header, height);
+		self.channel_manager.block_disconnected(header, height);
+		self.chain_monitor.block_disconnected(header, height);
+		self.output_sweeper.block_disconnected(header, height);
+	}
+}

src/chain/mod.rs

@@ -7,6 +7,7 @@
 
 mod bitcoind_rpc;
 
+use crate::chain::bitcoind_rpc::{BitcoindRpcClient, BoundedHeaderCache, ChainListener};
 use crate::config::{
 	Config, EsploraSyncConfig, BDK_CLIENT_CONCURRENCY, BDK_CLIENT_STOP_GAP,
 	BDK_WALLET_SYNC_TIMEOUT_SECS, FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, LDK_WALLET_SYNC_TIMEOUT_SECS,
@@ -22,11 +23,15 @@ use crate::logger::{log_bytes, log_error, log_info, log_trace, FilesystemLogger,
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
 
-use lightning::chain::{Confirm, Filter};
+use lightning::chain::{Confirm, Filter, Listen};
 use lightning::util::ser::Writeable;
 
 use lightning_transaction_sync::EsploraSyncClient;
 
+use lightning_block_sync::init::synchronize_listeners;
+use lightning_block_sync::poll::ChainPoller;
+use lightning_block_sync::SpvClient;
+
 use bdk_esplora::EsploraAsyncExt;
 
 use esplora_client::AsyncClient as EsploraAsyncClient;
@@ -37,14 +42,14 @@ use std::collections::HashMap;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
-use self::bitcoind_rpc::BitcoindRpcClient;
-
 // The default Esplora server we're using.
 pub(crate) const DEFAULT_ESPLORA_SERVER_URL: &str = "https://blockstream.info/api";
 
 // The default Esplora client timeout we're using.
 pub(crate) const DEFAULT_ESPLORA_CLIENT_TIMEOUT_SECS: u64 = 10;
 
+const CHAIN_POLLING_INTERVAL_SECS: u64 = 1;
+
 pub(crate) enum WalletSyncStatus {
 	Completed,
 	InProgress { subscribers: tokio::sync::broadcast::Sender<Result<(), Error>> },
@@ -241,7 +246,144 @@ impl ChainSource {
 					}
 				}
 			},
-			Self::BitcoindRpc { .. } => todo!(),
+			Self::BitcoindRpc {
+				bitcoind_rpc_client,
+				onchain_wallet,
+				kv_store,
+				config,
+				logger,
+				node_metrics,
+				..
+			} => {
+				let mut header_cache = BoundedHeaderCache::new();
+				let channel_manager_best_block_hash =
+					channel_manager.current_best_block().block_hash;
+				let sweeper_best_block_hash = output_sweeper.current_best_block().block_hash;
+				let onchain_wallet_best_block_hash = onchain_wallet.current_best_block().block_hash;
+
+				let mut chain_listeners = vec![
+					(
+						onchain_wallet_best_block_hash,
+						&**onchain_wallet as &(dyn Listen + Send + Sync),
+					),
+					(
+						channel_manager_best_block_hash,
+						&*channel_manager as &(dyn Listen + Send + Sync),
+					),
+					(sweeper_best_block_hash, &*output_sweeper as &(dyn Listen + Send + Sync)),
+				];
+
+				// TODO: Eventually we might want to see if we can synchronize `ChannelMonitor`s
+				// before giving them to `ChainMonitor` it the first place. However, this isn't
+				// trivial as we load them on initialization (in the `Builder`) and only gain
+				// network access during `start`. For now, we just make sure we get the worst known
+				// block hash and sychronize them via `ChainMonitor`.
+				if let Some(worst_channel_monitor_block_hash) = chain_monitor
+					.list_monitors()
+					.iter()
+					.flat_map(|(txo, _)| chain_monitor.get_monitor(*txo))
+					.map(|m| m.current_best_block())
+					.min_by_key(|b| b.height)
+					.map(|b| b.block_hash)
+				{
+					chain_listeners.push((
+						worst_channel_monitor_block_hash,
+						&*chain_monitor as &(dyn Listen + Send + Sync),
+					));
+				}
+
+				let chain_tip = loop {
+					match synchronize_listeners(
+						bitcoind_rpc_client.as_ref(),
+						config.network,
+						&mut header_cache,
+						chain_listeners.clone(),
+					)
+					.await
+					{
+						Ok(chain_tip) => {
+							{
+								let unix_time_secs_opt = SystemTime::now()
+									.duration_since(UNIX_EPOCH)
+									.ok()
+									.map(|d| d.as_secs());
+								let mut locked_node_metrics = node_metrics.write().unwrap();
+								locked_node_metrics.latest_lightning_wallet_sync_timestamp =
+									unix_time_secs_opt;
+								locked_node_metrics.latest_onchain_wallet_sync_timestamp =
+									unix_time_secs_opt;
+								write_node_metrics(
+									&*locked_node_metrics,
+									Arc::clone(&kv_store),
+									Arc::clone(&logger),
+								)
+								.unwrap_or_else(|e| {
+									log_error!(logger, "Failed to persist node metrics: {}", e);
+								});
+							}
+							break chain_tip;
+						},
+
+						Err(e) => {
+							log_error!(logger, "Failed to synchronize chain listeners: {:?}", e);
+							tokio::time::sleep(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS))
+								.await;
+						},
+					}
+				};
+
+				let chain_poller =
+					ChainPoller::new(Arc::clone(&bitcoind_rpc_client), config.network);
+				let chain_listener = ChainListener {
+					onchain_wallet: Arc::clone(&onchain_wallet),
+					channel_manager: Arc::clone(&channel_manager),
+					chain_monitor: Arc::clone(&chain_monitor),
+					output_sweeper: Arc::clone(&output_sweeper),
+				};
+				let mut spv_client =
+					SpvClient::new(chain_tip, chain_poller, &mut 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);
+
+				// Start the polling loop.
+				loop {
+					tokio::select! {
+						_ = stop_sync_receiver.changed() => {
+							log_trace!(
+								logger,
+								"Stopping polling for new chain data.",
+							);
+							return;
+						}
+						_ = chain_polling_interval.tick() => {
+							let _ = spv_client.poll_best_tip().await.map_err(|e| {
+								log_error!(logger, "Failed to poll for chain data: {:?}", e);
+							});
+							{
+								let unix_time_secs_opt = SystemTime::now()
+									.duration_since(UNIX_EPOCH)
+									.ok()
+									.map(|d| d.as_secs());
+								let mut locked_node_metrics = node_metrics.write().unwrap();
+								locked_node_metrics.latest_lightning_wallet_sync_timestamp =
+									unix_time_secs_opt;
+								locked_node_metrics.latest_onchain_wallet_sync_timestamp =
+									unix_time_secs_opt;
+								write_node_metrics(
+									&*locked_node_metrics,
+									Arc::clone(&kv_store),
+									Arc::clone(&logger),
+								)
+								.unwrap_or_else(|e| {
+									log_error!(logger, "Failed to persist node metrics: {}", e);
+								});
+							}
+						}
+					}
+				}
+			},
 		}
 	}
 

src/wallet/mod.rs

@@ -13,7 +13,7 @@ use crate::fee_estimator::{ConfirmationTarget, FeeEstimator};
 use crate::Error;
 
 use lightning::chain::chaininterface::BroadcasterInterface;
-use lightning::chain::Listen;
+use lightning::chain::{BestBlock, Listen};
 
 use lightning::events::bump_transaction::{Utxo, WalletSource};
 use lightning::ln::msgs::{DecodeError, UnsignedGossipMessage};
@@ -85,6 +85,11 @@ where
 		self.inner.lock().unwrap().start_sync_with_revealed_spks().build()
 	}
 
+	pub(crate) fn current_best_block(&self) -> BestBlock {
+		let checkpoint = self.inner.lock().unwrap().latest_checkpoint();
+		BestBlock { block_hash: checkpoint.hash(), height: checkpoint.height() }
+	}
+
 	pub(crate) fn apply_update(&self, update: impl Into<Update>) -> Result<(), Error> {
 		let mut locked_wallet = self.inner.lock().unwrap();
 		match locked_wallet.apply_update(update) {