f - Bound and deduplicate pending classification retries

LDK re-broadcasts pending claims every 30 seconds (and sweeps once per
block) until they confirm, so while the payment store is unavailable,
the list of pending retries accumulated a copy per rebroadcast —
memory, retry load on the struggling store, and a duplicate broadcast
burst on recovery all growing with the outage's duration.

A package whose transactions already await a retry is not queued
again, and the rest are bounded: at the bound, the oldest waiting
non-funding package is dropped to make room — its transactions return
with LDK's next periodic rebroadcast — but never a funding package,
whose transaction would be left confirming without a recorded
candidate. Fee-bumped rebroadcast variants carry new txids, so the
bound, not the dedup, is what limits their accumulation.

Implemented with Claude Code.

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

Author: jkczyz

src/chain/mod.rs

@@ -37,7 +37,7 @@ 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::tx_broadcaster::{BroadcastPackage, RetryQueue, ScheduleOutcome};
 use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet};
 use crate::{Error, PersistedNodeMetrics};
 
@@ -610,33 +610,49 @@ impl ChainSource {
 		// 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();
+		let mut retries = RetryQueue::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! {
+			let next_retry_at = retries.next_retry_at();
+			let package = tokio::select! {
 				_ = stop_tx_bcast_receiver.changed() => {
 					log_debug!(
-						tx_bcast_logger,
+						self.logger,
 						"Stopping broadcasting transactions.",
 					);
 					return;
 				}
-				Some(next_package) = receiver.recv() => {
-					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));
-					}
-				}
+				Some(next_package) = receiver.recv() => next_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));
-					}
+					retries.pop_next().expect("a retry is queued")
+				}
+			};
+			if let Err(package) = self.classify_and_broadcast(package).await {
+				let retry_at = tokio::time::Instant::now() + FAILED_CLASSIFY_RETRY_DELAY;
+				match retries.schedule(package, retry_at) {
+					ScheduleOutcome::Scheduled { dropped: None } => {},
+					ScheduleOutcome::Scheduled { dropped: Some(dropped) } => {
+						log_error!(
+							self.logger,
+							"Dropped the oldest package awaiting a classification retry; LDK re-broadcasts its transactions periodically: {:?}",
+							dropped.sorted_txids(),
+						);
+					},
+					ScheduleOutcome::AlreadyQueued(duplicate) => {
+						log_debug!(
+							self.logger,
+							"Dropped a re-broadcast package; an identical one already awaits a classification retry: {:?}",
+							duplicate.sorted_txids(),
+						);
+					},
+					ScheduleOutcome::Refused(package) => {
+						log_error!(
+							self.logger,
+							"Dropped a package failing classification; too many await retries: {:?}",
+							package.sorted_txids(),
+						);
+					},
 				}
 			}
 		}

src/tx_broadcaster.rs

@@ -5,21 +5,30 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use std::collections::VecDeque;
 use std::ops::Deref;
 use std::sync::{Mutex as StdMutex, Weak};
 
-use bitcoin::Transaction;
+use bitcoin::{Transaction, Txid};
 use lightning::chain::chaininterface::{
 	BroadcasterInterface, TransactionType as LdkTransactionType,
 };
 use tokio::sync::{mpsc, Mutex, MutexGuard};
+use tokio::time::Instant;
 
 use crate::logger::{log_error, LdkLogger};
 use crate::types::Wallet;
 use crate::Error;
 
 const BCAST_PACKAGE_QUEUE_SIZE: usize = 256;
 
+/// The most non-funding packages [`RetryQueue`] holds. Claims and sweeps re-enter the
+/// broadcast queue on LDK's periodic rebroadcast timers, so one dropped here resurfaces on its
+/// own once the store recovers. Funding packages don't count against the bound: nothing
+/// re-broadcasts them for us, and they are finite — one per negotiated candidate, since a copy
+/// of a waiting package is never queued twice.
+const MAX_QUEUED_RETRIES: usize = BCAST_PACKAGE_QUEUE_SIZE;
+
 /// 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
@@ -47,6 +56,100 @@ impl BroadcastPackage {
 		let txs = self.0.into_iter().map(|(tx, _)| tx).collect();
 		SortedTransactions::sort_parents_child_package_topologically(txs)
 	}
+
+	/// The packaged transactions' txids in sorted order, identifying the package's effect on
+	/// chain: two packages with the same txids broadcast the same transactions.
+	pub(crate) fn sorted_txids(&self) -> Vec<Txid> {
+		let mut txids: Vec<Txid> = self.0.iter().map(|(tx, _)| tx.compute_txid()).collect();
+		txids.sort_unstable();
+		txids
+	}
+
+	/// Whether the package contains a funding transaction (a channel open or splice), whose
+	/// classification writes the payment record tracking the funding.
+	fn contains_funding(&self) -> bool {
+		self.0.iter().any(|(_, tx_type)| {
+			matches!(
+				tx_type,
+				Some(
+					LdkTransactionType::Funding { .. }
+						| LdkTransactionType::InteractiveFunding { .. }
+				)
+			)
+		})
+	}
+}
+
+/// What [`RetryQueue::schedule`] did with a package, so the caller can log the cases in which
+/// the package won't be retried as-is.
+pub(crate) enum ScheduleOutcome {
+	/// The package waits for its retry deadline. When the bound was reached, the oldest waiting
+	/// non-funding package was dropped to make room and is returned — its transactions resurface
+	/// with LDK's next periodic rebroadcast.
+	Scheduled { dropped: Option<BroadcastPackage> },
+	/// A package broadcasting the same transactions already waits, and its retry covers this
+	/// one: the incoming package is dropped and returned.
+	AlreadyQueued(BroadcastPackage),
+	/// The bound was reached and every waiting package is a funding package, which must not be
+	/// dropped: the incoming package is refused and returned.
+	Refused(BroadcastPackage),
+}
+
+/// Packages whose classification failed, each waiting out a retry delay before its next attempt.
+/// Deduplicated and bounded: LDK re-broadcasts pending claims every 30 seconds (and sweeps once
+/// per block) until they confirm, so while the store is unavailable, copies would otherwise
+/// accumulate without bound and replay as a burst on recovery. An identical copy is never queued
+/// twice — the waiting entry and its deadline stand; fee-bumped rebroadcast variants carry new
+/// txids, so the bound — not the dedup — is what limits their accumulation.
+pub(crate) struct RetryQueue(VecDeque<(Instant, Vec<Txid>, BroadcastPackage)>);
+
+impl RetryQueue {
+	pub(crate) fn new() -> Self {
+		Self(VecDeque::new())
+	}
+
+	/// The deadline of the next retry, if a package is waiting. Packages are scheduled with a fixed
+	/// delay, so the front entry is always the next to retry.
+	pub(crate) fn next_retry_at(&self) -> Option<Instant> {
+		self.0.front().map(|(deadline, _, _)| *deadline)
+	}
+
+	/// Removes and returns the package scheduled to retry first.
+	pub(crate) fn pop_next(&mut self) -> Option<BroadcastPackage> {
+		self.0.pop_front().map(|(_, _, package)| package)
+	}
+
+	/// Schedules a package to retry at `retry_at`, unless a package with the same transactions already
+	/// waits or accepting it would exceed [`MAX_QUEUED_RETRIES`] with no non-funding package to
+	/// drop for it; see [`ScheduleOutcome`].
+	pub(crate) fn schedule(
+		&mut self, package: BroadcastPackage, retry_at: Instant,
+	) -> ScheduleOutcome {
+		let txids = package.sorted_txids();
+		if self.0.iter().any(|(_, waiting, _)| *waiting == txids) {
+			// Same transactions, same classification outcome: keep the waiting entry and its
+			// earlier deadline. The one same-txid package with a *different* type is LDK's
+			// re-typed generic-funding rebroadcast of a promoted 0conf splice, which always
+			// arrives after the interactive-funding original (the zero-conf rebroadcast canary
+			// tests assert that ordering), so the entry kept is the richer of the two — and its
+			// classification declines the downgrade anyway.
+			return ScheduleOutcome::AlreadyQueued(package);
+		}
+
+		let mut dropped = None;
+		if !package.contains_funding() && self.0.len() >= MAX_QUEUED_RETRIES {
+			// Drop the oldest non-funding package: LDK re-broadcasts its transactions
+			// periodically, while the incoming package may carry a fresher fee-bumped variant.
+			// A funding package is never dropped — nothing would re-broadcast it, and losing it
+			// leaves its transaction confirming without a recorded candidate.
+			match self.0.iter().position(|(_, _, waiting)| !waiting.contains_funding()) {
+				Some(oldest) => dropped = self.0.remove(oldest).map(|(_, _, package)| package),
+				None => return ScheduleOutcome::Refused(package),
+			}
+		}
+		self.0.push_back((retry_at, txids, package));
+		ScheduleOutcome::Scheduled { dropped }
+	}
 }
 
 pub(crate) struct SortedTransactions(Vec<Transaction>);
@@ -171,7 +274,10 @@ mod tests {
 	use bitcoin::hashes::Hash;
 	use bitcoin::{Amount, OutPoint, ScriptBuf, Sequence, Transaction, TxIn, TxOut, Txid, Witness};
 
-	use super::SortedTransactions;
+	use super::{
+		BroadcastPackage, LdkTransactionType, RetryQueue, ScheduleOutcome, SortedTransactions,
+		MAX_QUEUED_RETRIES,
+	};
 
 	fn txin(txid: Txid, vout: u32) -> TxIn {
 		TxIn {
@@ -312,4 +418,142 @@ mod tests {
 	fn topological_sort_accepts_empty_vec() {
 		SortedTransactions::sort_parents_child_package_topologically(Vec::new());
 	}
+
+	fn funding_package(tx: &Transaction) -> BroadcastPackage {
+		BroadcastPackage::new(&[(tx, LdkTransactionType::Funding { channels: vec![] })])
+	}
+
+	fn deadline(secs: u64) -> tokio::time::Instant {
+		tokio::time::Instant::now() + std::time::Duration::from_secs(secs)
+	}
+
+	/// A re-broadcast of the same transactions is not queued again: the waiting entry keeps its
+	/// earlier deadline and its package — the first arrival carries the richer classification
+	/// when LDK later re-types a rebroadcast.
+	#[tokio::test]
+	async fn retry_queue_queues_identical_transactions_once() {
+		let tx = parent_tx(1);
+		let mut retries = RetryQueue::new();
+
+		let first_deadline = deadline(2);
+		assert!(matches!(
+			retries.schedule(funding_package(&tx), first_deadline),
+			ScheduleOutcome::Scheduled { dropped: None }
+		));
+		assert!(matches!(
+			retries.schedule(BroadcastPackage::unclassified(tx.clone()), deadline(4)),
+			ScheduleOutcome::AlreadyQueued(_)
+		));
+
+		assert_eq!(retries.next_retry_at(), Some(first_deadline));
+		let kept = retries.pop_next().expect("the first package is kept");
+		assert!(
+			matches!(kept.transactions()[0].1, Some(LdkTransactionType::Funding { .. })),
+			"the first-scheduled package must be kept"
+		);
+		assert!(retries.pop_next().is_none());
+	}
+
+	#[tokio::test]
+	async fn retry_queue_retries_in_schedule_order() {
+		let (tx_a, tx_b) = (parent_tx(1), parent_tx(2));
+		let mut retries = RetryQueue::new();
+
+		assert!(matches!(
+			retries.schedule(BroadcastPackage::unclassified(tx_a.clone()), deadline(2)),
+			ScheduleOutcome::Scheduled { dropped: None }
+		));
+		assert!(matches!(
+			retries.schedule(BroadcastPackage::unclassified(tx_b.clone()), deadline(2)),
+			ScheduleOutcome::Scheduled { dropped: None }
+		));
+
+		let popped = retries.pop_next().expect("first package");
+		assert_eq!(popped.sorted_txids(), vec![tx_a.compute_txid()]);
+		let popped = retries.pop_next().expect("second package");
+		assert_eq!(popped.sorted_txids(), vec![tx_b.compute_txid()]);
+	}
+
+	/// Distinct transactions (e.g. fee-bumped claim variants during a store outage) are held to
+	/// the bound: the oldest non-funding package is dropped for an incoming one, never a funding
+	/// package.
+	#[tokio::test]
+	async fn retry_queue_drops_the_oldest_non_funding_package_at_the_bound() {
+		fn numbered_tx(n: u32) -> Transaction {
+			Transaction {
+				version: bitcoin::transaction::Version::TWO,
+				lock_time: bitcoin::absolute::LockTime::ZERO,
+				input: vec![txin(Txid::from_byte_array([7u8; 32]), n)],
+				output: vec![txout(1_000)],
+			}
+		}
+
+		let mut retries = RetryQueue::new();
+		let funding_tx = numbered_tx(0);
+		assert!(matches!(
+			retries.schedule(funding_package(&funding_tx), deadline(2)),
+			ScheduleOutcome::Scheduled { dropped: None }
+		));
+		let oldest_claim = numbered_tx(1);
+		for n in 1..(MAX_QUEUED_RETRIES as u32) {
+			assert!(matches!(
+				retries.schedule(BroadcastPackage::unclassified(numbered_tx(n)), deadline(2)),
+				ScheduleOutcome::Scheduled { dropped: None }
+			));
+		}
+
+		// At the bound, an incoming non-funding package drops the oldest waiting one — not the
+		// older funding package.
+		let new_claim = numbered_tx(MAX_QUEUED_RETRIES as u32);
+		match retries.schedule(BroadcastPackage::unclassified(new_claim.clone()), deadline(2)) {
+			ScheduleOutcome::Scheduled { dropped: Some(dropped) } => {
+				assert_eq!(dropped.sorted_txids(), vec![oldest_claim.compute_txid()]);
+			},
+			_ => panic!("the incoming claim must be scheduled by dropping the oldest one"),
+		}
+
+		// An incoming funding package is never dropped for the bound.
+		let new_funding_tx = numbered_tx(MAX_QUEUED_RETRIES as u32 + 1);
+		assert!(matches!(
+			retries.schedule(funding_package(&new_funding_tx), deadline(2)),
+			ScheduleOutcome::Scheduled { dropped: None }
+		));
+
+		let mut remaining = Vec::new();
+		while let Some(package) = retries.pop_next() {
+			remaining.extend(package.sorted_txids());
+		}
+		assert!(remaining.contains(&funding_tx.compute_txid()), "funding is never dropped");
+		assert!(remaining.contains(&new_claim.compute_txid()));
+		assert!(!remaining.contains(&oldest_claim.compute_txid()));
+	}
+
+	/// When only funding packages wait at the bound, an incoming non-funding package is refused:
+	/// LDK re-broadcasts claims and sweeps periodically, while a dropped funding package would
+	/// leave its transaction confirming without a recorded candidate.
+	#[tokio::test]
+	async fn retry_queue_refuses_a_non_funding_package_over_waiting_funding_packages() {
+		fn numbered_tx(n: u32) -> Transaction {
+			Transaction {
+				version: bitcoin::transaction::Version::TWO,
+				lock_time: bitcoin::absolute::LockTime::ZERO,
+				input: vec![txin(Txid::from_byte_array([8u8; 32]), n)],
+				output: vec![txout(1_000)],
+			}
+		}
+
+		let mut retries = RetryQueue::new();
+		for n in 0..(MAX_QUEUED_RETRIES as u32) {
+			assert!(matches!(
+				retries.schedule(funding_package(&numbered_tx(n)), deadline(2)),
+				ScheduleOutcome::Scheduled { dropped: None }
+			));
+		}
+
+		let claim = numbered_tx(MAX_QUEUED_RETRIES as u32);
+		assert!(matches!(
+			retries.schedule(BroadcastPackage::unclassified(claim), deadline(2)),
+			ScheduleOutcome::Refused(_)
+		));
+	}
 }