Fix three funding payment record bugs

src/chain/mod.rs

@@ -37,9 +37,15 @@ use crate::config::{BackgroundSyncConfig, Config, WALLET_SYNC_INTERVAL_MINIMUM_S
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::runtime::Runtime;
+use crate::tx_broadcaster::BroadcastPackage;
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
 
+/// How long to wait before re-classifying a package whose classification failed. Long enough to
+/// give a struggling store room to recover, short against the ~minutes until the transaction
+/// could confirm.
+const FAILED_CLASSIFY_RETRY_DELAY: Duration = Duration::from_secs(2);
+
 /// We use this parent-child TRUC package to make sure the configured chain source supports
 /// broadcasting packages via the `submitpackage` Bitcoin Core RPC.
 const PARENT_TXID: &str = "9a015f93fac6cb203c2b994e18b85176eb0354a22a468255516f3c6002d3f696";
@@ -562,12 +568,53 @@ impl ChainSource {
 		}
 	}
 
+	/// Classifies the package's funding broadcasts into payment records, then broadcasts it.
+	/// Returns the package back on classification failure so the caller can retry it after a
+	/// delay: broadcasting a tx we failed to record would leave it on-chain without a payment,
+	/// while dropping the package would not keep an interactively funded tx off-chain (the
+	/// counterparty broadcasts it regardless), only leave it confirming without a recorded
+	/// candidate.
+	async fn classify_and_broadcast(
+		&self, package: BroadcastPackage,
+	) -> Result<(), BroadcastPackage> {
+		if let Err(e) = self.tx_broadcaster.classify_package(&package).await {
+			log_error!(
+				self.logger,
+				"Delaying broadcast: failed to persist payment records, will retry: {:?}",
+				e,
+			);
+			return Err(package);
+		}
+		let package = package.into_sorted_transactions();
+		match &self.kind {
+			#[cfg(feature = "chain-esplora")]
+			ChainSourceKind::Esplora(esplora_chain_source) => {
+				esplora_chain_source.process_transaction_broadcast(package).await
+			},
+			#[cfg(feature = "chain-electrum")]
+			ChainSourceKind::Electrum(electrum_chain_source) => {
+				electrum_chain_source.process_transaction_broadcast(package).await
+			},
+			#[cfg(feature = "chain-bitcoind")]
+			ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
+				bitcoind_chain_source.process_transaction_broadcast(package).await
+			},
+		}
+		Ok(())
+	}
+
 	pub(crate) async fn continuously_process_broadcast_queue(
 		&self, mut stop_tx_bcast_receiver: tokio::sync::watch::Receiver<()>,
 	) {
 		let mut receiver = self.tx_broadcaster.get_broadcast_queue().await;
+		// Packages whose classification failed, each waiting out FAILED_CLASSIFY_RETRY_DELAY
+		// before its next attempt. New packages keep flowing while these wait, and pending
+		// retries die with the loop on shutdown rather than resurfacing after a later start.
+		let mut parked: Vec<(tokio::time::Instant, BroadcastPackage)> = Vec::new();
 		loop {
 			let tx_bcast_logger = Arc::clone(&self.logger);
+			// Entries are appended with a fixed delay, so the first is always the next due.
+			let next_retry_at = parked.first().map(|(deadline, _)| *deadline);
 			tokio::select! {
 				_ = stop_tx_bcast_receiver.changed() => {
 					log_debug!(
@@ -577,34 +624,18 @@ impl ChainSource {
 					return;
 				}
 				Some(next_package) = receiver.recv() => {
-					// Classify funding broadcasts into payment records before sending. If
-					// classification fails we skip the broadcast, since broadcasting a tx we
-					// failed to record would leave it on-chain without a payment.
-					let package = match self.tx_broadcaster.classify_package(next_package).await {
-						Ok(package) => package,
-						Err(e) => {
-							log_error!(
-								tx_bcast_logger,
-								"Skipping broadcast: failed to persist payment records: {:?}",
-								e,
-							);
-							continue;
-						},
-					};
-					let package = package.into_sorted_transactions();
-					match &self.kind {
-						#[cfg(feature = "chain-esplora")]
-						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_transaction_broadcast(package).await
-						},
-						#[cfg(feature = "chain-electrum")]
-						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_transaction_broadcast(package).await
-						},
-						#[cfg(feature = "chain-bitcoind")]
-						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_transaction_broadcast(package).await
-						},
+					if let Err(package) = self.classify_and_broadcast(next_package).await {
+						let retry_at = tokio::time::Instant::now() + FAILED_CLASSIFY_RETRY_DELAY;
+						parked.push((retry_at, package));
+					}
+				}
+				_ = tokio::time::sleep_until(
+					next_retry_at.unwrap_or_else(tokio::time::Instant::now)
+				), if next_retry_at.is_some() => {
+					let (_, package) = parked.remove(0);
+					if let Err(package) = self.classify_and_broadcast(package).await {
+						let retry_at = tokio::time::Instant::now() + FAILED_CLASSIFY_RETRY_DELAY;
+						parked.push((retry_at, package));
 					}
 				}
 			}

src/tx_broadcaster.rs

@@ -133,12 +133,10 @@ where
 		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 transaction on-chain without a record.
-	pub(crate) async fn classify_package(
-		&self, package: BroadcastPackage,
-	) -> Result<BroadcastPackage, Error> {
+	/// Classifies a queued package into payment records. Returns `Err` if any classification
+	/// fails; callers must not broadcast the package in that case, since a crash would leave the
+	/// transaction on-chain without a record — but must retry it later rather than drop it.
+	pub(crate) async fn classify_package(&self, package: &BroadcastPackage) -> Result<(), Error> {
 		let wallet_opt = self.wallet.lock().expect("lock").as_ref().and_then(Weak::upgrade);
 		if let Some(wallet) = wallet_opt {
 			for (tx, tx_type) in package.transactions() {
@@ -147,7 +145,7 @@ where
 				}
 			}
 		}
-		Ok(package)
+		Ok(())
 	}
 
 	pub(crate) fn broadcast_unclassified_transaction(&self, tx: Transaction) {