Persist payment transaction data without blocking LDK

Previously the BroadcasterInterface implementation wrote the payment
record synchronously when LDK invoked it. With a remote KV store this
could block LDK's message handling for hundreds of milliseconds per
call, noticeably during force-close bursts or splice broadcasts.

Persistence now happens asynchronously and must complete before the
transaction is sent to the chain client. If persistence fails, the
broadcast is dropped: a payment record must exist for every on-chain tx
we emit, otherwise a crash could leave the tx confirmed with no
matching record.

Generated with assistance from Claude Code.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jkczyz

src/chain/mod.rs

@@ -24,7 +24,7 @@ use crate::config::{
 	WALLET_SYNC_INTERVAL_MINIMUM_SECS,
 };
 use crate::fee_estimator::OnchainFeeEstimator;
-use crate::logger::{log_debug, log_info, log_trace, LdkLogger, Logger};
+use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::runtime::Runtime;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, NodeMetrics};
@@ -453,15 +453,26 @@ impl ChainSource {
 					return;
 				}
 				Some(next_package) = receiver.recv() => {
+					let txs = match self.tx_broadcaster.classify_package(next_package).await {
+						Ok(txs) => txs,
+						Err(e) => {
+							log_error!(
+								tx_bcast_logger,
+								"Skipping broadcast: failed to persist payment records: {:?}",
+								e,
+							);
+							continue;
+						},
+					};
 					match &self.kind {
 						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_broadcast_package(next_package).await
+							esplora_chain_source.process_broadcast_package(txs).await
 						},
 						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_broadcast_package(next_package).await
+							electrum_chain_source.process_broadcast_package(txs).await
 						},
 						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_broadcast_package(next_package).await
+							bitcoind_chain_source.process_broadcast_package(txs).await
 						},
 					}
 				}

src/tx_broadcaster.rs

@@ -14,20 +14,26 @@ use tokio::sync::{mpsc, Mutex, MutexGuard};
 
 use crate::logger::{log_error, LdkLogger};
 use crate::types::Wallet;
+use crate::Error;
 
 const BCAST_PACKAGE_QUEUE_SIZE: usize = 50;
 
+/// A package of transactions that LDK handed to the broadcaster in one
+/// `broadcast_transactions` call, along with each transaction's type. Queued until the
+/// background task classifies and broadcasts it.
+pub(crate) type BroadcastPackage = Vec<(Transaction, TransactionType)>;
+
 pub(crate) struct TransactionBroadcaster<L: Deref>
 where
 	L::Target: LdkLogger,
 {
-	queue_sender: mpsc::Sender<Vec<Transaction>>,
-	queue_receiver: Mutex<mpsc::Receiver<Vec<Transaction>>>,
+	queue_sender: mpsc::Sender<BroadcastPackage>,
+	queue_receiver: Mutex<mpsc::Receiver<BroadcastPackage>>,
 	/// Weak handle to the [`Wallet`] that performs classification of funding broadcasts
 	/// (channel opens and splices) into payment records. Remains `None` while the
-	/// builder is wiring the node up, during which broadcasts are still forwarded to
-	/// the queue but no payment record is written. [`Self::set_wallet`] installs the
-	/// handle once the [`Wallet`] exists.
+	/// builder is wiring the node up, during which broadcasts are forwarded to the
+	/// queue but no payment record is written. [`Self::set_wallet`] installs the handle
+	/// once the [`Wallet`] exists.
 	wallet: StdMutex<Option<Weak<Wallet>>>,
 	logger: L,
 }
@@ -55,30 +61,41 @@ where
 
 	pub(crate) async fn get_broadcast_queue(
 		&self,
-	) -> MutexGuard<'_, mpsc::Receiver<Vec<Transaction>>> {
+	) -> MutexGuard<'_, mpsc::Receiver<BroadcastPackage>> {
 		self.queue_receiver.lock().await
 	}
+
+	/// Classifies a queued package into payment records and returns the raw
+	/// transactions ready for the chain client. Returns `Err` if any classification
+	/// fails; callers must not broadcast the package in that case, since a crash would
+	/// leave the tx on-chain without a record.
+	pub(crate) async fn classify_package(
+		&self, package: BroadcastPackage,
+	) -> Result<Vec<Transaction>, Error> {
+		let wallet_opt = self.wallet.lock().expect("lock").as_ref().and_then(Weak::upgrade);
+		if let Some(wallet) = wallet_opt {
+			let package = tokio::task::spawn_blocking(move || {
+				for (tx, tx_type) in &package {
+					wallet.classify_broadcast(tx, tx_type)?;
+				}
+				Ok::<_, Error>(package)
+			})
+			.await
+			.map_err(|_| Error::PersistenceFailed)??;
+			Ok(package.into_iter().map(|(tx, _)| tx).collect())
+		} else {
+			Ok(package.into_iter().map(|(tx, _)| tx).collect())
+		}
+	}
 }
 
 impl<L: Deref> BroadcasterInterface for TransactionBroadcaster<L>
 where
 	L::Target: LdkLogger,
 {
 	fn broadcast_transactions(&self, txs: &[(&Transaction, TransactionType)]) {
-		let wallet = self.wallet.lock().expect("lock").as_ref().and_then(Weak::upgrade);
-		if let Some(wallet) = wallet {
-			for (tx, tx_type) in txs {
-				if let Err(e) = wallet.classify_broadcast(tx, tx_type) {
-					log_error!(
-						self.logger,
-						"Failed to classify broadcast tx {}: {:?}",
-						tx.compute_txid(),
-						e,
-					);
-				}
-			}
-		}
-		let package = txs.iter().map(|(t, _)| (*t).clone()).collect::<Vec<Transaction>>();
+		let package: BroadcastPackage =
+			txs.iter().map(|(tx, tx_type)| ((*tx).clone(), tx_type.clone())).collect();
 		self.queue_sender.try_send(package).unwrap_or_else(|e| {
 			log_error!(self.logger, "Failed to broadcast transactions: {}", e);
 		});