f - Keep classification retries inside the broadcast loop

The retry for a failed classification was a detached tokio::spawn that
outlived the node. Its comment claimed a re-send after shutdown would
fail because the queue had closed, but the queue receiver lives in the
broadcaster and is only dropped with the node, so the re-send succeeded
and a stale package would be classified and broadcast after a
stop()/start() cycle.

Park failed packages inside the broadcast loop instead and retry them
from a timer branch of the same select. New packages keep flowing while
a retry waits, and pending retries are dropped when the loop stops.

Implemented with Claude Code.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jkczyz

src/chain/mod.rs

@@ -26,9 +26,15 @@ use crate::config::{
 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";
@@ -520,12 +526,50 @@ 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 {
+			ChainSourceKind::Esplora(esplora_chain_source) => {
+				esplora_chain_source.process_transaction_broadcast(package).await
+			},
+			ChainSourceKind::Electrum(electrum_chain_source) => {
+				electrum_chain_source.process_transaction_broadcast(package).await
+			},
+			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!(
@@ -535,32 +579,18 @@ impl ChainSource {
 					return;
 				}
 				Some(next_package) = receiver.recv() => {
-					// Classify funding broadcasts into payment records before sending. If
-					// classification fails we delay the broadcast and retry, since 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.
-					if let Err(e) = self.tx_broadcaster.classify_package(&next_package).await {
-						log_error!(
-							tx_bcast_logger,
-							"Delaying broadcast: failed to persist payment records, will retry: {:?}",
-							e,
-						);
-						self.tx_broadcaster.requeue_failed_classify(next_package);
-						continue;
+					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));
 					}
-					let package = next_package.into_sorted_transactions();
-					match &self.kind {
-						ChainSourceKind::Esplora(esplora_chain_source) => {
-							esplora_chain_source.process_transaction_broadcast(package).await
-						},
-						ChainSourceKind::Electrum(electrum_chain_source) => {
-							electrum_chain_source.process_transaction_broadcast(package).await
-						},
-						ChainSourceKind::Bitcoind(bitcoind_chain_source) => {
-							bitcoind_chain_source.process_transaction_broadcast(package).await
-						},
+				}
+				_ = 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

@@ -7,7 +7,6 @@
 
 use std::ops::Deref;
 use std::sync::{Mutex as StdMutex, Weak};
-use std::time::Duration;
 
 use bitcoin::Transaction;
 use lightning::chain::chaininterface::{
@@ -21,11 +20,6 @@ use crate::Error;
 
 const BCAST_PACKAGE_QUEUE_SIZE: usize = 256;
 
-/// 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);
-
 /// 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. Built only via [`BroadcastPackage::new`] from such a call, so unrelated
@@ -141,8 +135,7 @@ where
 
 	/// 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 requeue it via
-	/// [`Self::requeue_failed_classify`] rather than drop it.
+	/// 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 {
@@ -155,20 +148,6 @@ where
 		Ok(())
 	}
 
-	/// Re-sends a package whose classification failed back into the queue after a delay, so a
-	/// transient persistence failure delays the broadcast instead of dropping the package.
-	/// Dropping an interactive-funding package would not even keep its transaction off-chain —
-	/// the counterparty broadcasts it regardless — it would only leave the transaction
-	/// confirming without a recorded candidate. If the queue has closed by the time the delay
-	/// elapses, the node is shutting down and the package is dropped with it.
-	pub(crate) fn requeue_failed_classify(&self, package: BroadcastPackage) {
-		let sender = self.queue_sender.clone();
-		tokio::spawn(async move {
-			tokio::time::sleep(FAILED_CLASSIFY_RETRY_DELAY).await;
-			let _ = sender.send(package).await;
-		});
-	}
-
 	pub(crate) fn broadcast_unclassified_transaction(&self, tx: Transaction) {
 		self.queue_sender.try_send(BroadcastPackage::unclassified(tx)).unwrap_or_else(|e| {
 			log_error!(self.logger, "Failed to broadcast transactions: {}", e);

src/wallet/mod.rs

@@ -4318,6 +4318,88 @@ mod tests {
 		loop_task.await.unwrap();
 	}
 
+	/// A package awaiting a classification retry must die when the node stops. When the retry
+	/// was a detached task, it outlived the broadcast loop: its re-send into the still-open
+	/// queue succeeded after `stop()`, so a later `start()` would classify and broadcast the
+	/// stale package.
+	#[tokio::test]
+	async fn failed_classification_retry_dies_at_stop() {
+		use lightning::chain::chaininterface::BroadcasterInterface;
+
+		let fail_store = FailSwitchStore::new();
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(fail_store.clone()));
+		let wallet = new_test_wallet(Arc::clone(&store), false).await;
+		wallet.broadcaster.set_wallet(Arc::downgrade(&wallet));
+
+		let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
+		let chain_source = Arc::clone(&wallet.chain_source);
+		let loop_task = tokio::spawn(async move {
+			chain_source.continuously_process_broadcast_queue(stop_receiver).await
+		});
+
+		let script_pubkey = wallet
+			.inner
+			.lock()
+			.unwrap()
+			.reveal_next_address(KeychainKind::External)
+			.address
+			.script_pubkey();
+		let tx = Transaction {
+			version: bitcoin::transaction::Version::TWO,
+			lock_time: LockTime::ZERO,
+			input: Vec::new(),
+			output: vec![TxOut { value: Amount::from_sat(10_000), script_pubkey }],
+		};
+		let counterparty_node_id = PublicKey::from_str(
+			"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798",
+		)
+		.unwrap();
+
+		// Queue the broadcast while payment persistence is failing and wait for the loop to
+		// fail a classification attempt, leaving a retry pending.
+		fail_store.fail_writes.store(true, Ordering::Release);
+		wallet.broadcaster.broadcast_transactions(&[(
+			&tx,
+			LdkTransactionType::Funding {
+				channels: vec![(counterparty_node_id, ChannelId([7u8; 32]))],
+			},
+		)]);
+		let mut failed_writes = 0;
+		for _ in 0..100 {
+			tokio::time::sleep(Duration::from_millis(100)).await;
+			failed_writes = fail_store.failed_writes.load(Ordering::Acquire);
+			if failed_writes > 0 {
+				break;
+			}
+		}
+		assert!(failed_writes > 0, "classification never attempted a payment-store write");
+
+		// Stop the node with the retry still pending, then bring the loop back up with
+		// working persistence, as a stop()/start() cycle would.
+		stop_sender.send(()).unwrap();
+		loop_task.await.unwrap();
+		fail_store.fail_writes.store(false, Ordering::Release);
+
+		let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
+		let chain_source = Arc::clone(&wallet.chain_source);
+		let loop_task = tokio::spawn(async move {
+			chain_source.continuously_process_broadcast_queue(stop_receiver).await
+		});
+
+		// Watch well past the retry delay: the package from before the stop must not be
+		// classified or broadcast by the restarted loop.
+		for _ in 0..40 {
+			tokio::time::sleep(Duration::from_millis(100)).await;
+			assert!(
+				wallet.payment_store.list_page(None).await.unwrap().objects.is_empty(),
+				"a package from before stop() resurfaced after restart"
+			);
+		}
+
+		stop_sender.send(()).unwrap();
+		loop_task.await.unwrap();
+	}
+
 	/// Barrier test, classification-first ordering: wallet sync's confirmation handling must
 	/// wait for classification's two-store write pair. Classification is parked between its
 	/// payment-store and pending-store writes (the torn window) and only then is the