Implement syncing via bip157 (kyoto)

Add a storage of watched scripts
Add a mechanism to process the filters and watched scripts
Add last synced checkpoint, channel to watch its updates
Make possible the kyoto node to start/stop, maintain the original peer list provided.
Add header cache for not to redownload them when applying.
Add wait_until_synced
Add tx broadcasting.
Add tests.

Author: randomlogin

Cargo.toml

@@ -76,7 +76,7 @@ serde_json = { version = "1.0.128", default-features = false, features = ["std"]
 log = { version = "0.4.22", default-features = false, features = ["std"]}
 
 async-trait = { version = "0.1", default-features = false }
-bip157 = "0.3.4"
+bip157 = "0.4.2"
 vss-client = { package = "vss-client-ng", version = "0.5" }
 prost = { version = "0.11.6", default-features = false}
 #bitcoin-payment-instructions = { version = "0.6" }

src/chain/kyoto.rs

@@ -14,7 +14,7 @@ use std::collections::HashMap;
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
-use bitcoin::{Script, Txid};
+use bitcoin::{Script, ScriptBuf, Txid};
 use lightning::chain::{BestBlock, Filter};
 
 use crate::chain::bitcoind::{BitcoindChainSource, UtxoSourceClient};
@@ -104,7 +104,7 @@ enum ChainSourceKind {
 	Esplora(EsploraChainSource),
 	Electrum(ElectrumChainSource),
 	Bitcoind(BitcoindChainSource),
-    Kyoto(KyotoChainSource),
+	Kyoto(KyotoChainSource),
 }
 
 impl ChainSource {
@@ -221,9 +221,7 @@ impl ChainSource {
 			ChainSourceKind::Electrum(electrum_chain_source) => {
 				electrum_chain_source.start(runtime)?
 			},
-			ChainSourceKind::Kyoto(kyoto_chain_source) => {
-				kyoto_chain_source.start(runtime)?
-			},
+			ChainSourceKind::Kyoto(kyoto_chain_source) => kyoto_chain_source.start(runtime)?,
 			_ => {
 				// Nothing to do for other chain sources.
 			},
@@ -254,6 +252,16 @@ impl ChainSource {
 		self.registered_txids.lock().unwrap().clone()
 	}
 
+	/// Register a script pubkey to watch in compact block filters.
+	///
+	/// This is a no-op for chain backends other than BIP157/kyoto, which rely on transaction-based
+	/// sync and don't need an explicit script watchlist.
+	pub(crate) fn register_script(&self, script: ScriptBuf) {
+		if let ChainSourceKind::Kyoto(kyoto_chain_source) = &self.kind {
+			kyoto_chain_source.register_output(script);
+		}
+	}
+
 	pub(crate) fn is_transaction_based(&self) -> bool {
 		match &self.kind {
 			ChainSourceKind::Esplora(_) => true,
@@ -399,8 +407,7 @@ impl ChainSource {
 		}
 	}
 
-	// Synchronize the onchain wallet via transaction-based protocols (i.e., Esplora, Electrum,
-	// etc.)
+	// Synchronize the onchain wallet via transaction-based protocols (i.e., Esplora, Electrum, etc.)
 	pub(crate) async fn sync_onchain_wallet(
 		&self, onchain_wallet: Arc<Wallet>,
 	) -> Result<(), Error> {
@@ -450,7 +457,7 @@ impl ChainSource {
 		}
 	}
 
-	pub(crate) async fn poll_and_update_listeners(
+	pub(crate) async fn sync_listeners_to_tip(
 		&self, onchain_wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>,
 		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
 	) -> Result<(), Error> {
@@ -475,8 +482,8 @@ impl ChainSource {
 					)
 					.await
 			},
-			ChainSourceKind::Kyoto { .. } => {
-				unreachable!("Listeners will be synced via the kyoto event loop")
+			ChainSourceKind::Kyoto(kyoto_chain_source) => {
+				kyoto_chain_source.wait_until_synced().await
 			},
 		}
 	}
@@ -544,7 +551,9 @@ impl Filter for ChainSource {
 				electrum_chain_source.register_tx(txid, script_pubkey)
 			},
 			ChainSourceKind::Bitcoind { .. } => (),
-		ChainSourceKind::Kyoto { .. } => (),
+			ChainSourceKind::Kyoto(kyoto_chain_source) => {
+				kyoto_chain_source.register_output(script_pubkey.to_owned())
+			},
 		}
 	}
 	fn register_output(&self, output: lightning::chain::WatchedOutput) {
@@ -556,7 +565,9 @@ impl Filter for ChainSource {
 				electrum_chain_source.register_output(output)
 			},
 			ChainSourceKind::Bitcoind { .. } => (),
-		ChainSourceKind::Kyoto { .. } => (),
+			ChainSourceKind::Kyoto(kyoto_chain_source) => {
+				kyoto_chain_source.register_output(output.script_pubkey)
+			},
 		}
 	}
 }

src/chain/mod.rs

@@ -272,6 +272,9 @@ impl Node {
 		let chain_source = Arc::clone(&self.chain_source);
 		self.runtime.block_on(async move { chain_source.update_fee_rate_estimates().await })?;
 
+		// For BIP157 backends: seed kyoto's script watchlist with all previously-derived wallet addresses
+		self.wallet.seed_watched_scripts();
+
 		// Spawn background task continuously syncing onchain, lightning, and fee rate cache.
 		let stop_sync_receiver = self.stop_sender.subscribe();
 		let chain_source = Arc::clone(&self.chain_source);
@@ -1642,7 +1645,7 @@ impl Node {
 			} else {
 				chain_source.update_fee_rate_estimates().await?;
 				chain_source
-					.poll_and_update_listeners(
+					.sync_listeners_to_tip(
 						sync_wallet,
 						sync_cman,
 						sync_cmon,

src/lib.rs

@@ -9,6 +9,7 @@ use std::future::Future;
 use std::ops::Deref;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use bdk_chain::spk_client::{FullScanRequest, SyncRequest};
 use bdk_wallet::descriptor::ExtendedDescriptor;
@@ -122,6 +123,35 @@ impl Wallet {
 		self.inner.lock().unwrap().start_sync_with_revealed_spks().build()
 	}
 
+	/// Register all previously-revealed wallet scripts into the chain source's script watchlist.
+	///
+	/// For BIP157 backends the kyoto compact-filter scanner only checks scripts that have been
+	/// explicitly registered. When the node restarts, previously-derived addresses need to be
+	/// re-seeded so that blocks containing payments to those addresses are not silently skipped.
+	/// This is a no-op for non-BIP157 backends.
+	pub(crate) fn seed_watched_scripts(&self) {
+		let locked_wallet = self.inner.lock().unwrap();
+		for ((_, _), spk) in locked_wallet.spk_index().revealed_spks(..) {
+			self.chain_source.register_script(spk);
+		}
+	}
+
+	/// Advance the wallet's internal [`LocalChain`] to include the given block header.
+	///
+	/// Called by the BIP157 backend for every `ChainUpdate::Connected` event so the chain stays
+	/// in sync with kyoto's header stream. Without this, subsequent `block_connected` calls would
+	/// fail because BDK requires the parent block to already be in the chain before connecting the
+	/// next block.
+	pub(crate) fn apply_header(&self, header: &bitcoin::block::Header, height: u32) {
+		let mut locked_wallet = self.inner.lock().unwrap();
+		let block_id = bdk_chain::BlockId { height, hash: header.block_hash() };
+		let latest_cp = locked_wallet.latest_checkpoint().insert(block_id);
+		let update = bdk_wallet::Update { chain: Some(latest_cp), ..Default::default() };
+		if let Err(e) = locked_wallet.apply_update(update) {
+			log_error!(self.logger, "BIP157: Failed to apply header at height {}: {}", height, e);
+		}
+	}
+
 	pub(crate) fn get_cached_txs(&self) -> Vec<Arc<Transaction>> {
 		self.inner.lock().unwrap().tx_graph().full_txs().map(|tx_node| tx_node.tx).collect()
 	}
@@ -329,13 +359,17 @@ impl Wallet {
 							let tx_refs: Vec<(
 								&Transaction,
 								lightning::chain::chaininterface::TransactionType,
-							)> =
-								txs_to_broadcast
-									.iter()
-									.map(|tx| {
-										(tx, lightning::chain::chaininterface::TransactionType::Sweep { channels: vec![] })
-									})
-									.collect();
+							)> = txs_to_broadcast
+								.iter()
+								.map(|tx| {
+									(
+										tx,
+										lightning::chain::chaininterface::TransactionType::Sweep {
+											channels: vec![],
+										},
+									)
+								})
+								.collect();
 							self.broadcaster.broadcast_transactions(&tx_refs);
 							log_info!(
 								self.logger,
@@ -426,44 +460,50 @@ impl Wallet {
 	) -> Result<Transaction, Error> {
 		let fee_rate = self.fee_estimator.estimate_fee_rate(confirmation_target);
 
-		let mut locked_wallet = self.inner.lock().unwrap();
-		let mut tx_builder = locked_wallet.build_tx();
+		let tx = {
+			let mut locked_wallet = self.inner.lock().unwrap();
+			let mut tx_builder = locked_wallet.build_tx();
 
-		tx_builder.add_recipient(output_script, amount).fee_rate(fee_rate).nlocktime(locktime);
+			tx_builder.add_recipient(output_script, amount).fee_rate(fee_rate).nlocktime(locktime);
 
-		let mut psbt = match tx_builder.finish() {
-			Ok(psbt) => {
-				log_trace!(self.logger, "Created funding PSBT: {:?}", psbt);
-				psbt
-			},
-			Err(err) => {
-				log_error!(self.logger, "Failed to create funding transaction: {}", err);
-				return Err(err.into());
-			},
-		};
+			let mut psbt = match tx_builder.finish() {
+				Ok(psbt) => {
+					log_trace!(self.logger, "Created funding PSBT: {:?}", psbt);
+					psbt
+				},
+				Err(err) => {
+					log_error!(self.logger, "Failed to create funding transaction: {}", err);
+					return Err(err.into());
+				},
+			};
 
-		match locked_wallet.sign(&mut psbt, SignOptions::default()) {
-			Ok(finalized) => {
-				if !finalized {
-					return Err(Error::OnchainTxCreationFailed);
-				}
-			},
-			Err(err) => {
-				log_error!(self.logger, "Failed to create funding transaction: {}", err);
-				return Err(err.into());
-			},
-		}
+			match locked_wallet.sign(&mut psbt, SignOptions::default()) {
+				Ok(finalized) => {
+					if !finalized {
+						return Err(Error::OnchainTxCreationFailed);
+					}
+				},
+				Err(err) => {
+					log_error!(self.logger, "Failed to create funding transaction: {}", err);
+					return Err(err.into());
+				},
+			}
 
-		let mut locked_persister = self.persister.lock().unwrap();
-		locked_wallet.persist(&mut locked_persister).map_err(|e| {
-			log_error!(self.logger, "Failed to persist wallet: {}", e);
-			Error::PersistenceFailed
-		})?;
+			let mut locked_persister = self.persister.lock().unwrap();
+			locked_wallet.persist(&mut locked_persister).map_err(|e| {
+				log_error!(self.logger, "Failed to persist wallet: {}", e);
+				Error::PersistenceFailed
+			})?;
 
-		let tx = psbt.extract_tx().map_err(|e| {
-			log_error!(self.logger, "Failed to extract transaction: {}", e);
-			e
-		})?;
+			psbt.extract_tx().map_err(|e| {
+				log_error!(self.logger, "Failed to extract transaction: {}", e);
+				e
+			})?
+		}; // locked_wallet drops here
+
+		// Re-seed watched scripts so the BIP-157 chain source learns about any
+		// change addresses that BDK derived internally when building the tx.
+		self.seed_watched_scripts();
 
 		Ok(tx)
 	}
@@ -473,6 +513,8 @@ impl Wallet {
 		let mut locked_persister = self.persister.lock().unwrap();
 
 		let address_info = locked_wallet.reveal_next_address(KeychainKind::External);
+		// For BIP157 backends, register the script so kyoto watches it in compact filters.
+		self.chain_source.register_script(address_info.address.script_pubkey());
 		locked_wallet.persist(&mut locked_persister).map_err(|e| {
 			log_error!(self.logger, "Failed to persist wallet: {}", e);
 			Error::PersistenceFailed
@@ -485,6 +527,8 @@ impl Wallet {
 		let mut locked_persister = self.persister.lock().unwrap();
 
 		let address_info = locked_wallet.next_unused_address(KeychainKind::Internal);
+		// For BIP157 backends, register the script so kyoto watches it in compact filters.
+		self.chain_source.register_script(address_info.address.script_pubkey());
 		locked_wallet.persist(&mut locked_persister).map_err(|e| {
 			log_error!(self.logger, "Failed to persist wallet: {}", e);
 			Error::PersistenceFailed
@@ -811,7 +855,8 @@ impl Wallet {
 					let (sent, received) = locked_wallet.sent_and_received(&psbt.unsigned_tx);
 					let drain_amount = sent - received;
 					if spendable_amount_sats < drain_amount.to_sat() {
-						log_error!(self.logger,
+						log_error!(
+							self.logger,
 							"Unable to send payment due to insufficient funds. Available: {}sats, Required: {}",
 							spendable_amount_sats,
 							drain_amount,
@@ -846,13 +891,30 @@ impl Wallet {
 			})?
 		};
 
+		// Re-seed watched scripts so the BIP-157 chain source learns about any
+		// change addresses that BDK derived internally when building the tx.
+		self.seed_watched_scripts();
+
 		self.broadcaster.broadcast_transactions(&[(
 			&tx,
 			lightning::chain::chaininterface::TransactionType::Sweep { channels: vec![] },
 		)]);
 
 		let txid = tx.compute_txid();
 
+		// Eagerly register the just-sent tx as unconfirmed in BDK so the payment
+		// store entry is immediately visible via Node::payment(). Without this,
+		// backends without mempool support (e.g. BIP157) would only see the payment
+		// after the tx is confirmed in a block.
+		let now_secs = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
+		if let Err(e) = self.apply_mempool_txs(vec![(tx, now_secs)], vec![]) {
+			log_error!(
+				self.logger,
+				"Failed to eagerly apply outgoing transaction to payment store: {:?}",
+				e
+			);
+		}
+
 		match send_amount {
 			OnchainSendAmount::ExactRetainingReserve { amount_sats, .. } => {
 				log_info!(
@@ -1336,7 +1398,13 @@ impl Wallet {
 						final_fee_rate_sat_per_kwu.saturating_mul(3).saturating_div(2),
 					);
 					if required_fee_rate > max_allowed_fee_rate {
-						log_error!( self.logger, "BDK required fee rate {} exceeds sanity cap {} (1.5x our estimate) for tx {}", required_fee_rate, max_allowed_fee_rate, txid );
+						log_error!(
+							self.logger,
+							"BDK required fee rate {} exceeds sanity cap {} (1.5x our estimate) for tx {}",
+							required_fee_rate,
+							max_allowed_fee_rate,
+							txid
+						);
 						return Err(Error::InvalidFeeRate);
 					}