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/bitcoind.rs

@@ -568,16 +568,18 @@ impl BitcoindChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
 		// While it's a bit unclear when we'd be able to lean on Bitcoin Core >v28
 		// features, we should eventually switch to use `submitpackage` via the
 		// `rust-bitcoind-json-rpc` crate rather than just broadcasting individual
 		// transactions.
-		for tx in &package {
+		for tx in txs {
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(DEFAULT_TX_BROADCAST_TIMEOUT_SECS),
-				self.api_client.broadcast_transaction(tx),
+				self.api_client.broadcast_transaction(&tx),
 			);
 			match timeout_fut.await {
 				Ok(res) => match res {

src/chain/electrum.rs

@@ -275,7 +275,9 @@ impl ElectrumChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
 		let electrum_client: Arc<ElectrumRuntimeClient> = if let Some(client) =
 			self.electrum_runtime_status.read().expect("lock").client().as_ref()
 		{
@@ -285,7 +287,7 @@ impl ElectrumChainSource {
 			return;
 		};
 
-		for tx in package {
+		for tx in txs {
 			electrum_client.broadcast(tx).await;
 		}
 	}

src/chain/esplora.rs

@@ -352,12 +352,14 @@ impl EsploraChainSource {
 		Ok(())
 	}
 
-	pub(crate) async fn process_broadcast_package(&self, package: Vec<Transaction>) {
-		for tx in &package {
+	pub(crate) async fn process_broadcast_package(
+		&self, txs: impl IntoIterator<Item = Transaction>,
+	) {
+		for tx in txs {
 			let txid = tx.compute_txid();
 			let timeout_fut = tokio::time::timeout(
 				Duration::from_secs(self.sync_config.timeouts_config.tx_broadcast_timeout_secs),
-				self.esplora_client.broadcast(tx),
+				self.esplora_client.broadcast(&tx),
 			);
 			match timeout_fut.await {
 				Ok(res) => match res {

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,27 @@ impl ChainSource {
 					return;
 				}
 				Some(next_package) = receiver.recv() => {
+					let package = match self.tx_broadcaster.classify_package(next_package).await {
+						Ok(p) => p,
+						Err(e) => {
+							log_error!(
+								tx_bcast_logger,
+								"Skipping broadcast: failed to persist payment records: {:?}",
+								e,
+							);
+							continue;
+						},
+					};
+					let txs = package.into_iter().map(|(tx, _)| tx);
 					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,40 @@ 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 package 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<BroadcastPackage, Error> {
+		let wallet_opt = self.wallet.lock().expect("lock").as_ref().and_then(Weak::upgrade);
+		if let Some(wallet) = wallet_opt {
+			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)?
+		} else {
+			Ok(package)
+		}
+	}
 }
 
 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);
 		});