From c7e81230c6edc84fedc76d530f24a476047ee3e7 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 18 Aug 2026 14:27:39 -0500 Subject: [PATCH 1/8] Only adopt a funding payment's own transactions from wallet sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wallet sync resolves a funding payment's id for any transaction linked to the record through its conflicting txids, and then adopted that transaction's txid and confirmation outright. A cooperative close conflicts with a pending splice in exactly that way: the splice record would report the close's txid and confirmation under its InteractiveFunding type and contribution figures and graduate as if the splice had confirmed, while the close's own record never received its confirmation. Adopt a transaction only when it is part of the payment's funding history — the record's current txid or a classified candidate. Anything else is recorded under its own txid-keyed id, which also delivers the close's confirmation to the close's own record. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 189 ++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 164 insertions(+), 25 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index b9c12b4a7..f8dd13db1 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -346,12 +346,12 @@ impl Wallet { // duplicating) the record classification just wrote. let guard = self.funding_payment_update_lock.lock().await; - let payment_id = self + let mut payment_id = self .find_payment_by_txid(txid) .await? .unwrap_or_else(|| PaymentId(txid.to_byte_array())); - if self + match self .apply_funding_status_update_locked( &guard, payment_id, @@ -360,7 +360,13 @@ impl Wallet { ) .await? { - continue; + FundingStatusUpdate::Applied => continue, + FundingStatusUpdate::NotFunding => {}, + // Not part of the funding payment's history (e.g. a close spending the + // funding outpoint): record it under its own id below instead. + FundingStatusUpdate::Foreign => { + payment_id = PaymentId(txid.to_byte_array()); + }, } let payment = { @@ -487,12 +493,12 @@ impl Wallet { // with classification. let guard = self.funding_payment_update_lock.lock().await; - let payment_id = self + let mut payment_id = self .find_payment_by_txid(txid) .await? .unwrap_or_else(|| PaymentId(txid.to_byte_array())); - if self + match self .apply_funding_status_update_locked( &guard, payment_id, @@ -501,7 +507,13 @@ impl Wallet { ) .await? { - continue; + FundingStatusUpdate::Applied => continue, + FundingStatusUpdate::NotFunding => {}, + // Not part of the funding payment's history (e.g. a close spending the + // funding outpoint): record it under its own id below instead. + FundingStatusUpdate::Foreign => { + payment_id = PaymentId(txid.to_byte_array()); + }, } let payment = { @@ -563,12 +575,12 @@ impl Wallet { // with classification. let guard = self.funding_payment_update_lock.lock().await; - let payment_id = self + let mut payment_id = self .find_payment_by_txid(txid) .await? .unwrap_or_else(|| PaymentId(txid.to_byte_array())); - if self + match self .apply_funding_status_update_locked( &guard, payment_id, @@ -577,7 +589,13 @@ impl Wallet { ) .await? { - continue; + FundingStatusUpdate::Applied => continue, + FundingStatusUpdate::NotFunding => {}, + // Not part of the funding payment's history (e.g. a close spending the + // funding outpoint): record it under its own id below instead. + FundingStatusUpdate::Foreign => { + payment_id = PaymentId(txid.to_byte_array()); + }, } let payment = { @@ -1938,9 +1956,11 @@ impl Wallet { /// If `payment_id` refers to a classified funding payment, refreshes its confirmation status /// and the candidate txid the event refers to, while preserving the contribution-derived /// amount/fee and `tx_type` that wallet sync must not recompute from its own view: the wallet's - /// `sent`/`received` don't capture our contribution to a shared funding output. Returns `true` - /// when it handled the payment, so the caller skips the default on-chain path. Graduation to - /// `Succeeded` is left to `ChainTipChanged` after `ANTI_REORG_DELAY`. + /// `sent`/`received` don't capture our contribution to a shared funding output. Returns + /// [`FundingStatusUpdate::Applied`] when it handled the payment, so the caller skips the + /// default on-chain path — or [`FundingStatusUpdate::Foreign`] when the transaction is not + /// part of the payment's funding history, so the caller records it under its own id. + /// Graduation to `Succeeded` is left to `ChainTipChanged` after `ANTI_REORG_DELAY`. /// /// The caller must hold [`Self::funding_payment_update_lock`] — from resolving `payment_id` /// through its own last write, not just across this call — so that classification's two-store @@ -1949,38 +1969,51 @@ impl Wallet { async fn apply_funding_status_update_locked( &self, _guard: &tokio::sync::MutexGuard<'_, ()>, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus, - ) -> Result { + ) -> Result { // The caller's wallet-level lock keeps the candidate history stable while we await its - // read. The funding-type gate and write then share the payment store's mutation lock: - // against a separate payment `get`, a classification merging in between would have its - // `tx_type` and contribution figures clobbered by this stale snapshot. + // read. The funding-type gate, the candidate lookup, and the write then share the payment + // store's mutation lock: against a separate payment `get`, a classification merging in + // between would have its `tx_type` and contribution figures clobbered by this stale + // snapshot. let pending_payment = self.pending_payment_store.get(&payment_id).await?; + let mut outcome = FundingStatusUpdate::NotFunding; let mut handled = None; self.payment_store .mutate(&payment_id, |existing| { let payment = existing?; - let tx_type = match &payment.kind { + let (current_txid, tx_type) = match &payment.kind { PaymentKind::Onchain { + txid, tx_type: tx_type @ Some( TransactionType::Funding { .. } | TransactionType::InteractiveFunding { .. }, ), .. - } => tx_type.clone(), + } => (*txid, tx_type.clone()), _ => return None, }; + // Adopt the event's txid only when the transaction is part of this payment's + // funding history: its current txid or a classified candidate. A conflicting + // transaction that is neither — a close also spends the funding outpoint — must + // not overwrite the record. + let owns_event_tx = event_txid == current_txid + || pending_payment.as_ref().is_some_and(|p| p.candidate(event_txid).is_some()); + if !owns_event_tx { + outcome = FundingStatusUpdate::Foreign; + return None; + } // Report the figures of the candidate that actually confirmed, which need not be // the last one broadcast (an earlier, lower-fee candidate may win) and may carry // no figures at all (`None`) for a round we didn't contribute to. (`direction` is // invariant across a splice's candidates and cannot be changed through the store // anyway.) let mut target = payment.clone(); - if let Some(pending) = pending_payment.as_ref() { - if let Some(candidate) = pending.candidate(event_txid) { - target.amount_msat = candidate.amount_msat; - target.fee_paid_msat = candidate.fee_paid_msat; - } + if let Some(candidate) = + pending_payment.as_ref().and_then(|p| p.candidate(event_txid)) + { + target.amount_msat = candidate.amount_msat; + target.fee_paid_msat = candidate.fee_paid_msat; } target.kind = PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type }; @@ -1998,7 +2031,7 @@ impl Wallet { }) .await?; let Some(payment) = handled else { - return Ok(false); + return Ok(outcome); }; // Mirror the refreshed confirmation status onto the pending entry: `ChainTipChanged` // graduates by reading the pending entry's details, so it must see the new status. This is @@ -2008,7 +2041,7 @@ impl Wallet { let pending = self.create_pending_payment_from_tx(payment, Vec::new()); self.pending_payment_store.insert_or_update(pending).await?; } - Ok(true) + Ok(FundingStatusUpdate::Applied) } #[allow(deprecated)] @@ -2311,6 +2344,20 @@ fn aggregate_local_stakes(candidate: &FundingCandidate) -> LocalStakeAggregate { } } +/// The outcome of [`Wallet::apply_funding_status_update_locked`]. +enum FundingStatusUpdate { + /// The event's transaction belongs to the funding payment; its refreshed confirmation status + /// was applied (or was already current). + Applied, + /// The resolved payment is not a classified funding payment; the caller's default on-chain + /// handling applies under the resolved id. + NotFunding, + /// The event's transaction is not part of the funding payment's history — e.g. a close + /// spending the same funding outpoint — so the funding record must not adopt it; the caller + /// should record the transaction under its own txid-derived id. + Foreign, +} + impl Listen for Wallet { fn filtered_block_connected( &self, _header: &bitcoin::block::Header, @@ -3960,6 +4007,98 @@ mod tests { assert_eq!(wallet.find_payment_by_txid(txid2).await.unwrap(), Some(payment_id)); } + /// A cooperative close conflicts with a pending splice's funding transaction — both spend the + /// pre-splice funding outpoint — so sync records the close among the splice record's + /// conflicting txids, and the close's confirmation then resolves to the splice's PaymentId. + /// The funding record must not adopt the close's txid and confirmation as its own: the close + /// is not a round of the splice. It must land on a record keyed by the close's own id. + #[tokio::test] + async fn funding_record_does_not_adopt_a_conflicting_close() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let funding_outpoint = + bitcoin::OutPoint { txid: Txid::from_byte_array([3u8; 32]), vout: 0 }; + + // The close pays the shutdown script, which is a wallet address. + let script_pubkey = wallet + .inner + .lock() + .unwrap() + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(); + let close_tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: vec![bitcoin::TxIn { + previous_output: funding_outpoint, + script_sig: bitcoin::ScriptBuf::new(), + sequence: bitcoin::Sequence::MAX, + witness: bitcoin::Witness::new(), + }], + output: vec![TxOut { value: Amount::from_sat(90_000), script_pubkey }], + }; + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + + // Sync saw the close double-spend the splice's funding transaction. + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![close_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + let event = WalletEvent::TxConfirmed { + txid: close_txid, + tx: Arc::new(close_tx), + block_time: confirmed_block_time(5), + old_block_time: None, + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let funding = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + match &funding.kind { + PaymentKind::Onchain { txid, status, tx_type } => { + assert_eq!(*txid, splice_txid, "the record must not adopt the close's txid"); + assert!(matches!(status, ConfirmationStatus::Unconfirmed)); + assert!(matches!(tx_type, Some(TransactionType::InteractiveFunding { .. }))); + }, + kind => panic!("unexpected kind {:?}", kind), + } + assert_eq!(funding.amount_msat, Some(1_000_000)); + assert_eq!(funding.fee_paid_msat, Some(500)); + + let close = wallet + .payment_store + .get(&PaymentId(close_txid.to_byte_array())) + .await + .unwrap() + .unwrap(); + match &close.kind { + PaymentKind::Onchain { txid, status, .. } => { + assert_eq!(*txid, close_txid); + assert!(matches!(status, ConfirmationStatus::Confirmed { .. })); + }, + kind => panic!("unexpected kind {:?}", kind), + } + } + /// A funding-typed broadcast that doesn't touch the on-chain wallet must not be recorded. /// LDK re-broadcasts a promoted-but-unconfirmed 0conf splice through its generic funding /// path, so a splice the interactive-funding classification deliberately declined — no local From 970eb1aeb2b0b517ed074288fa2ccbedaf803f61 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 18 Aug 2026 17:07:12 -0500 Subject: [PATCH 2/8] Retry funding-broadcast classification instead of dropping it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A queued broadcast whose payment-record classification failed was dropped outright, on the theory that broadcasting a transaction we failed to record would leave it on-chain without a payment. For interactive funding that theory doesn't hold: the counterparty broadcasts the same transaction once the signature exchange completes, so dropping the package keeps nothing off-chain — it only guarantees the round is never recorded as a candidate on our side. The funding-status ownership gate then treats the round's confirmation as foreign to the funding record and re-keys it to a stray duplicate record, which shadows the funding record's txid lookups permanently: the splice payment stays Pending forever while an untyped duplicate holds the confirmation. Keep the package alive instead: requeue it after a short delay and retry classification until it succeeds, holding the broadcast back the whole time. Classification failures are persistence failures, so the retry is unbounded — a store that never recovers keeps the node from functioning anyway — and every failed round is logged. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/chain/mod.rs | 29 ++++++++-------- src/tx_broadcaster.rs | 33 ++++++++++++++---- src/wallet/mod.rs | 81 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 21 deletions(-) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index f01c1c8cb..22151e9ec 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -578,20 +578,21 @@ impl ChainSource { } 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(); + // 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; + } + let package = next_package.into_sorted_transactions(); match &self.kind { #[cfg(feature = "chain-esplora")] ChainSourceKind::Esplora(esplora_chain_source) => { diff --git a/src/tx_broadcaster.rs b/src/tx_broadcaster.rs index 782112dad..248926d45 100644 --- a/src/tx_broadcaster.rs +++ b/src/tx_broadcaster.rs @@ -7,6 +7,7 @@ use std::ops::Deref; use std::sync::{Mutex as StdMutex, Weak}; +use std::time::Duration; use bitcoin::Transaction; use lightning::chain::chaininterface::{ @@ -20,6 +21,11 @@ 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 @@ -133,12 +139,11 @@ 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 { + /// 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. + 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 +152,21 @@ where } } } - Ok(package) + 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) { diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8dd13db1..6aa1300c8 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -4241,6 +4241,87 @@ mod tests { assert_unchanged(&wallet, payment_id, true).await; } + /// A funding broadcast whose classification fails must be retried, not dropped: for + /// interactive funding the counterparty broadcasts the same transaction regardless of + /// whether we do, so dropping the package permanently leaves the confirming transaction + /// unrecorded as a candidate — and the funding-status ownership gate then routes its + /// confirmation to a stray duplicate record instead of the funding record. + #[tokio::test] + async fn failed_funding_classification_is_retried_not_dropped() { + use lightning::chain::chaininterface::BroadcasterInterface; + + let fail_store = FailSwitchStore::new(); + let store: Arc = Arc::new(DynStoreWrapper(fail_store.clone())); + let wallet = new_test_wallet(Arc::clone(&store), false).await; + wallet.broadcaster.set_wallet(Arc::downgrade(&wallet)); + + // Run the production broadcast-queue loop. The broadcast itself fails fast against the + // fixture's unroutable Esplora server, which is irrelevant here: the record is written + // during classification, before the broadcast attempt. + 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 + }); + + // A funding transaction paying the wallet passes the wallet-activity guard, so its + // classification reaches the payment-store write. + 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. + fail_store.fail_writes.store(true, Ordering::Release); + wallet.broadcaster.broadcast_transactions(&[( + &tx, + LdkTransactionType::Funding { + channels: vec![(counterparty_node_id, ChannelId([7u8; 32]))], + }, + )]); + + // Let the loop fail at least one classification round; a failed classification must not + // leave a partial record behind. + tokio::time::sleep(Duration::from_secs(3)).await; + assert!(wallet.payment_store.list_filter(|_| true).is_empty()); + + // Once writes recover, the package must still be alive to classify. + fail_store.fail_writes.store(false, Ordering::Release); + let mut recorded = Vec::new(); + for _ in 0..100 { + tokio::time::sleep(Duration::from_millis(100)).await; + recorded = wallet.payment_store.list_filter(|_| true); + if !recorded.is_empty() { + break; + } + } + assert!( + !recorded.is_empty(), + "the failed classification was never retried; the package was dropped" + ); + assert_eq!(recorded.len(), 1); + assert!(matches!( + recorded[0].kind, + PaymentKind::Onchain { tx_type: Some(TransactionType::Funding { .. }), .. } + )); + + 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 From 3b64639bcc6dd6d7ad669bd26b8f51cd4dea5c55 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 26 Aug 2026 16:21:30 -0500 Subject: [PATCH 3/8] f - Wait for a failed write before letting the retry test recover The retry test slept a fixed three seconds and assumed classification had failed by then; if writes were re-enabled before the first attempt, the test would pass without any retry happening. Count failed writes in FailSwitchStore and wait for one before re-enabling writes. Also fix the test's store reads to use list_page: the payment store's cache is bounded, so list_filter is unavailable, and this commit did not compile its tests standalone (the conversion had landed in the following commit). Implemented with Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 29 ++++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 6aa1300c8..63c8719ae 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -2734,7 +2734,7 @@ fn funding_reclassification_update( #[cfg(all(test, any(feature = "chain-esplora", feature = "chain-electrum")))] mod tests { - use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use bdk_chain::{BlockId, ConfirmationBlockTime}; @@ -2764,11 +2764,13 @@ mod tests { const EXTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)"; const INTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)"; - /// An in-memory store whose writes can be made to fail on demand. + /// An in-memory store whose writes can be made to fail on demand, counting the failures so + /// tests can wait for a write to have actually failed rather than guessing with a sleep. #[derive(Clone)] struct FailSwitchStore { inner: Arc, fail_writes: Arc, + failed_writes: Arc, } impl FailSwitchStore { @@ -2776,6 +2778,7 @@ mod tests { Self { inner: Arc::new(InMemoryStore::new()), fail_writes: Arc::new(AtomicBool::new(false)), + failed_writes: Arc::new(AtomicUsize::new(0)), } } } @@ -2792,11 +2795,13 @@ mod tests { ) -> impl Future> + 'static + Send { let inner = Arc::clone(&self.inner); let fail_writes = Arc::clone(&self.fail_writes); + let failed_writes = Arc::clone(&self.failed_writes); let primary_namespace = primary_namespace.to_string(); let secondary_namespace = secondary_namespace.to_string(); let key = key.to_string(); async move { if fail_writes.load(Ordering::Acquire) { + failed_writes.fetch_add(1, Ordering::AcqRel); return Err(io::Error::new(io::ErrorKind::Other, "writes disabled")); } KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await @@ -4293,17 +4298,27 @@ mod tests { }, )]); - // Let the loop fail at least one classification round; a failed classification must not - // leave a partial record behind. - tokio::time::sleep(Duration::from_secs(3)).await; - assert!(wallet.payment_store.list_filter(|_| true).is_empty()); + // Wait until the loop has actually failed a classification write; re-enabling writes + // before the first attempt would let the first attempt succeed and the test pass + // without any retry happening. A failed classification must not leave a partial + // record behind. + 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"); + assert!(wallet.payment_store.list_page(None).await.unwrap().objects.is_empty()); // Once writes recover, the package must still be alive to classify. fail_store.fail_writes.store(false, Ordering::Release); let mut recorded = Vec::new(); for _ in 0..100 { tokio::time::sleep(Duration::from_millis(100)).await; - recorded = wallet.payment_store.list_filter(|_| true); + recorded = wallet.payment_store.list_page(None).await.unwrap().objects; if !recorded.is_empty() { break; } From 1f406a6f9c2eaf9155a593f9c9eb8f4c3b455d91 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 26 Aug 2026 16:25:44 -0500 Subject: [PATCH 4/8] 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. Queue 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 --- src/chain/mod.rs | 86 +++++++++++++++++++++++++++++-------------- src/tx_broadcaster.rs | 23 +----------- src/wallet/mod.rs | 82 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 50 deletions(-) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 22151e9ec..947378d7e 100644 --- a/src/chain/mod.rs +++ b/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,35 +624,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 { - #[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 - }, + } + _ = 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)); } } } diff --git a/src/tx_broadcaster.rs b/src/tx_broadcaster.rs index 248926d45..c40592558 100644 --- a/src/tx_broadcaster.rs +++ b/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); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 63c8719ae..9790e4f4c 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -4337,6 +4337,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 = 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 From 3a8eb128e1665282d0aec88479e25a0df7983ac5 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 2 Sep 2026 11:05:07 -0500 Subject: [PATCH 5/8] f - Keep stale classification retries from reverting the record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A queued classification can retry after a newer candidate of the same funding already classified. The retry carries the candidate history as of its own broadcast, so applying it rotated the record's txid back to the older candidate and shrank the stored candidate history — after which wallet sync could no longer map the newer transaction to the record and would file it as a foreign duplicate. A fresh interactive-funding classification always carries the record's current txid in its history, so one that doesn't is stale: ignore it, and never let a candidate-history update drop stored candidates. Implemented with Claude Code. Co-Authored-By: Claude Fable 5 --- src/payment/pending_payment_store.rs | 59 ++++++- src/wallet/mod.rs | 231 +++++++++++++++++++++++++-- 2 files changed, 271 insertions(+), 19 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 30a113537..df893b661 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -105,9 +105,19 @@ impl StorableObject for PendingPaymentDetails { updated |= self.conflicting_txids.len() != conflicts_len; } - // Each classify passes the complete candidate history, so a non-empty update replaces the - // stored list. An empty update (e.g. a non-funding payment) leaves it untouched. - if !update.candidates.is_empty() && self.candidates != update.candidates { + // Each classify passes the candidate history as of its own broadcast, so a non-empty + // update replaces the stored list. An empty update (e.g. a non-funding payment) leaves it + // untouched — as does an update missing a stored candidate: the history only ever grows, + // so such an update was built before that candidate existed (a classification retry + // running after a newer round classified) and replacing would orphan the newer round's + // transactions. + let extends_history = |stored: &FundingTxCandidate| { + update.candidates.iter().any(|candidate| candidate.txid == stored.txid) + }; + if !update.candidates.is_empty() + && self.candidates != update.candidates + && self.candidates.iter().all(extends_history) + { self.candidates = update.candidates; updated = true; } @@ -243,6 +253,49 @@ mod tests { ); } + /// The candidate history only ever grows. An update carrying a shorter history was built + /// before the newer candidates existed — a classification retry running after a newer round + /// classified — and must not shrink the stored list, or the newer candidates' transactions + /// could no longer be mapped back to the record. + #[test] + fn candidate_history_never_shrinks() { + let txid_a = test_txid(1); + let txid_b = test_txid(2); + let txid_c = test_txid(3); + let payment_id = PaymentId(txid_a.to_byte_array()); + let candidate = |txid, fee| FundingTxCandidate { + txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(fee), + }; + let history = vec![candidate(txid_a, 400), candidate(txid_b, 500)]; + + let mut pending = PendingPaymentDetails::new( + pending_onchain_payment(payment_id, txid_b), + Vec::new(), + history.clone(), + ); + let stale_update = PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: None, + candidates: vec![candidate(txid_a, 400)], + }; + assert!(!pending.update(stale_update), "a stale history must not shrink the stored one"); + assert_eq!(pending.candidates, history); + + // A history that extends the stored one still replaces it, refreshed figures included. + let extended = vec![candidate(txid_a, 400), candidate(txid_b, 550), candidate(txid_c, 600)]; + let fresh_update = PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: None, + candidates: extended.clone(), + }; + assert!(pending.update(fresh_update)); + assert_eq!(pending.candidates, extended); + } + #[test] fn funding_classification_pending_update_preserves_mirrored_confirmation() { use bitcoin::BlockHash; diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 9790e4f4c..c8477c9d7 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1831,11 +1831,22 @@ impl Wallet { // The record was written above and payment records are never removed, so absence // means the write failed out; fall back to the fresh details. let recorded = payment_store.get(&id).await?.unwrap_or(details); + // A candidate history that lacks the record's current txid is stale — a queued + // classification retrying after a newer round classified. The merge arm below + // refuses such a history; recreating a missing entry from it would smuggle it + // past that refusal, so leave the recreation to a fresh classification (the + // newer round's own write, or its retry) instead. + let stale = match &recorded.kind { + PaymentKind::Onchain { txid, .. } if !candidates.is_empty() => { + !candidates.iter().any(|c| c.txid == *txid) + }, + _ => false, + }; Ok(match existing { // The inserted entry embeds the post-write record rather than the fresh // details, so a confirmation wallet sync already recorded keeps driving // graduation. - None if recorded.status == PaymentStatus::Pending => { + None if recorded.status == PaymentStatus::Pending && !stale => { Some(PendingPaymentDetails::new(recorded, Vec::new(), candidates)) }, // The payment already advanced beyond Pending: the graduation path removed @@ -2714,6 +2725,29 @@ fn funding_reclassification_update( return PaymentDetailsUpdate::new(details.id); } + // An interactive-funding classification carries the full candidate history as of its own + // broadcast, and once a record is funding-classified its txid only ever names a candidate + // from that history. A classification whose history lacks such a record's current txid was + // therefore built before that candidate existed — a queued retry running after a newer round + // classified. Applying it would rotate the record backwards; the newer round's + // classification already recorded everything this one knows. A record that is not yet + // funding-classified gives no such signal — wallet sync can have rotated its txid to a + // conflicting transaction that is no candidate at all — so its first classification must + // still land. + if !candidates.is_empty() { + if let Some(PaymentKind::Onchain { + txid: current_txid, + tx_type: + Some(TransactionType::Funding { .. } | TransactionType::InteractiveFunding { .. }), + .. + }) = current.map(|payment| &payment.kind) + { + if !candidates.iter().any(|c| c.txid == *current_txid) { + return PaymentDetailsUpdate::new(details.id); + } + } + } + let mut update = PaymentDetailsUpdate::funding_reclassification(details); if let Some(PaymentKind::Onchain { txid: confirmed_txid, @@ -3807,12 +3841,20 @@ mod tests { #[test] fn funding_reclassification_update_keeps_the_active_candidate() { + let prior_txid = Txid::from_byte_array([1u8; 32]); let active_txid = Txid::from_byte_array([2u8; 32]); - let candidates = vec![FundingTxCandidate { - txid: active_txid, - amount_msat: Some(1_000_000), - fee_paid_msat: Some(500), - }]; + let candidates = vec![ + FundingTxCandidate { + txid: prior_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(400), + }, + FundingTxCandidate { + txid: active_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + ]; let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed); // No record yet: the update describes the active candidate. @@ -3820,25 +3862,79 @@ mod tests { assert_eq!(update.txid, Some(active_txid)); assert_eq!(update.amount_msat, Some(Some(1_000_000))); - // An unconfirmed record: still the active candidate (RBF rotation). - let unconfirmed = - onchain_details(Txid::from_byte_array([1u8; 32]), ConfirmationStatus::Unconfirmed); + // An unconfirmed record on the prior candidate: rotate to the active one (RBF). + let unconfirmed = onchain_details(prior_txid, ConfirmationStatus::Unconfirmed); let update = funding_reclassification_update(details.clone(), &candidates, Some(&unconfirmed)); assert_eq!(update.txid, Some(active_txid)); // The record confirmed the active candidate itself: nothing to substitute. let current = onchain_details(active_txid, confirmed_status()); - let update = funding_reclassification_update(details.clone(), &candidates, Some(¤t)); + let update = funding_reclassification_update(details, &candidates, Some(¤t)); assert_eq!(update.txid, Some(active_txid)); assert_eq!(update.amount_msat, Some(Some(1_000_000))); + } - // A confirmed txid outside the candidate history (e.g. the record is an unrelated - // same-id payment): fall back to the active candidate; `PaymentDetails::update` keeps - // the confirmed figures in place on mismatch. - let foreign = onchain_details(Txid::from_byte_array([9u8; 32]), confirmed_status()); - let update = funding_reclassification_update(details, &candidates, Some(&foreign)); - assert_eq!(update.txid, Some(active_txid)); + /// A classification whose candidate history lacks a funding-classified record's current txid + /// was built before that candidate existed — a queued retry running after a newer round + /// classified — and must move nothing, whatever the record's confirmation state. A record + /// that is not yet funding-classified gives no such signal (wallet sync can have rotated its + /// txid to a conflicting non-candidate), so its first classification must still land. + #[test] + fn funding_reclassification_update_refuses_a_stale_candidate_history() { + let stale_txid = Txid::from_byte_array([1u8; 32]); + let newer_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId(stale_txid.to_byte_array()); + let stale_history = vec![FundingTxCandidate { + txid: stale_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(400), + }]; + let details = + interactive_funding_details(payment_id, stale_txid, Some(1_000_000), Some(400)); + + // The record moved on to a newer candidate while this classification was queued. + let unconfirmed = + interactive_funding_details(payment_id, newer_txid, Some(1_000_000), Some(500)); + let update = + funding_reclassification_update(details.clone(), &stale_history, Some(&unconfirmed)); + let mut updated = unconfirmed.clone(); + assert!(!updated.update(update), "a stale retry must not move an unconfirmed record"); + assert_eq!(updated, unconfirmed); + + // Same when the newer candidate has already confirmed. + let mut confirmed = unconfirmed.clone(); + confirmed.kind = PaymentKind::Onchain { + txid: newer_txid, + status: confirmed_status(), + tx_type: Some(TransactionType::InteractiveFunding { channels: vec![] }), + }; + let update = + funding_reclassification_update(details.clone(), &stale_history, Some(&confirmed)); + let mut updated = confirmed.clone(); + assert!(!updated.update(update), "a stale retry must not move a confirmed record"); + assert_eq!(updated, confirmed); + + // A record that was never funding-classified: wallet sync rotated its txid to a + // conflicting transaction, which is no candidate. Its first classification is not stale + // and must land. + let mut unclassified = + interactive_funding_details(payment_id, newer_txid, Some(1_000_000), Some(500)); + unclassified.kind = PaymentKind::Onchain { + txid: newer_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + }; + let update = funding_reclassification_update(details, &stale_history, Some(&unclassified)); + let mut updated = unclassified.clone(); + assert!(updated.update(update), "a first classification must not be treated as stale"); + match &updated.kind { + PaymentKind::Onchain { txid, tx_type, .. } => { + assert_eq!(*txid, stale_txid); + assert!(matches!(tx_type, Some(TransactionType::InteractiveFunding { .. }))); + }, + kind => panic!("unexpected kind {:?}", kind), + } } /// A funding-typed (re)classification of a record already classified as interactive funding @@ -4419,6 +4515,109 @@ mod tests { loop_task.await.unwrap(); } + /// A queued classification can retry after a newer candidate of the same funding already + /// classified: the retry carries the candidate history as of its own broadcast, which no + /// longer includes the newer candidate. Applying it would rotate the record's txid backwards + /// and shrink the stored candidate history, after which the newer transaction can no longer + /// be mapped back to the record and wallet sync would file it as a foreign duplicate. + #[tokio::test] + async fn stale_classification_retry_keeps_the_newer_candidate() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let txid_a = Txid::from_byte_array([1u8; 32]); + let txid_b = Txid::from_byte_array([2u8; 32]); + // The record's id is anchored to the first negotiated candidate, so the stale retry + // resolves to the same record. + let payment_id = PaymentId(txid_a.to_byte_array()); + let candidate_a = FundingTxCandidate { + txid: txid_a, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }; + let candidate_b = FundingTxCandidate { + txid: txid_b, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(999), + }; + + // The bump candidate B classifies first, carrying the full history [A, B]. + let fresh = interactive_funding_details(payment_id, txid_b, Some(1_000_000), Some(999)); + wallet + .persist_funding_payment(fresh, vec![candidate_a.clone(), candidate_b.clone()]) + .await + .unwrap(); + + // The queued classification of A retries, carrying the history as of A's broadcast. + let stale = interactive_funding_details(payment_id, txid_a, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(stale, vec![candidate_a.clone()]).await.unwrap(); + + let record = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + match &record.kind { + PaymentKind::Onchain { txid, .. } => { + assert_eq!(*txid, txid_b, "the stale retry must not rotate the record back"); + }, + kind => panic!("unexpected kind {:?}", kind), + } + assert_eq!(record.fee_paid_msat, Some(999)); + + let pending = wallet.pending_payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!( + pending.candidates, + vec![candidate_a, candidate_b], + "the stale retry must not shrink the candidate history" + ); + + // The consequence the history protects against: B must stay mapped to the record, or + // wallet sync would file it as a foreign duplicate. + assert_eq!(wallet.find_payment_by_txid(txid_b).await.unwrap(), Some(payment_id)); + } + + /// A missing pending entry is normally recreated from the incoming classification — but not + /// from a stale retry, whose truncated candidate history would otherwise slip past the merge + /// path's refusal. Recreation is left to a fresh classification instead. + #[tokio::test] + async fn stale_classification_retry_does_not_recreate_the_pending_entry() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let txid_a = Txid::from_byte_array([1u8; 32]); + let txid_b = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId(txid_a.to_byte_array()); + let candidate_a = FundingTxCandidate { + txid: txid_a, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }; + let candidate_b = FundingTxCandidate { + txid: txid_b, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(999), + }; + + // The newer round B classified, but its write pair was torn by the same store failure + // that queued this retry: the record exists, the pending entry does not. + let recorded = interactive_funding_details(payment_id, txid_b, Some(1_000_000), Some(999)); + wallet.payment_store.insert(recorded).await.unwrap(); + + // The queued classification of A retries with its pre-B history. + let stale = interactive_funding_details(payment_id, txid_a, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(stale, vec![candidate_a.clone()]).await.unwrap(); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none(), + "a stale retry must not recreate the pending entry from its truncated history" + ); + + // B's own retry recreates the entry with the full history. + let fresh = interactive_funding_details(payment_id, txid_b, Some(1_000_000), Some(999)); + wallet + .persist_funding_payment(fresh, vec![candidate_a.clone(), candidate_b.clone()]) + .await + .unwrap(); + let pending = wallet.pending_payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(pending.candidates, vec![candidate_a, candidate_b]); + } + /// 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 From 6cf748ad01671ddffbe8deb95fd954ee9f0a2941 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 2 Sep 2026 11:18:10 -0500 Subject: [PATCH 6/8] f - Bound and deduplicate pending classification retries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/chain/mod.rs | 52 ++++++--- src/tx_broadcaster.rs | 248 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 280 insertions(+), 20 deletions(-) diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 947378d7e..2d53cf9d6 100644 --- a/src/chain/mod.rs +++ b/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(), + ); + }, } } } diff --git a/src/tx_broadcaster.rs b/src/tx_broadcaster.rs index c40592558..da283dd5d 100644 --- a/src/tx_broadcaster.rs +++ b/src/tx_broadcaster.rs @@ -5,14 +5,16 @@ // 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; @@ -20,6 +22,13 @@ 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 { + let mut txids: Vec = 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 }, + /// 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, 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 { + self.0.front().map(|(deadline, _, _)| *deadline) + } + + /// Removes and returns the package scheduled to retry first. + pub(crate) fn pop_next(&mut self) -> Option { + 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); @@ -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(_) + )); + } } From 238ccc8acff9deb12e14f6931d2be80d3a2d307c Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Fri, 4 Sep 2026 10:51:44 -0500 Subject: [PATCH 7/8] f - Never drop a queued cooperative close for the retry bound --- src/tx_broadcaster.rs | 188 +++++++++++++++++++++++++++++++++++------- 1 file changed, 158 insertions(+), 30 deletions(-) diff --git a/src/tx_broadcaster.rs b/src/tx_broadcaster.rs index da283dd5d..3e5b846da 100644 --- a/src/tx_broadcaster.rs +++ b/src/tx_broadcaster.rs @@ -22,11 +22,11 @@ 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. +/// The most droppable 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. Packages nothing re-broadcasts — fundings and cooperative closes — +/// don't count against the bound: they are finite — one per negotiated funding candidate and +/// one per closing channel, 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` @@ -65,17 +65,30 @@ impl BroadcastPackage { 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 { .. } - ) - ) + /// Whether the package may be dropped to keep [`RetryQueue`] within its bound: every + /// transaction in it is re-broadcast by its originator, so a dropped package resurfaces on + /// its own. LDK re-hands claims, anchor bumps, and force-close commitments to the + /// broadcaster periodically, and the sweeper regenerates sweeps once per block. Nothing + /// re-broadcasts a funding transaction (a channel open or splice, whose classification + /// writes the payment record tracking the funding) or a cooperative close (whose channel is + /// gone from the `ChannelManager` by broadcast time), so a package containing either is + /// never dropped. + fn is_droppable(&self) -> bool { + self.0.iter().all(|(_, tx_type)| match tx_type { + Some( + LdkTransactionType::Funding { .. } + | LdkTransactionType::InteractiveFunding { .. } + | LdkTransactionType::CooperativeClose { .. }, + ) => false, + Some( + LdkTransactionType::UnilateralClose { .. } + | LdkTransactionType::AnchorBump { .. } + | LdkTransactionType::Claim { .. } + | LdkTransactionType::Sweep { .. }, + ) => true, + // Wallet-originated: re-submitted on chain tip changes. Never queued anyway, since + // classification of an untyped package is a no-op that can't fail. + None => true, }) } } @@ -84,14 +97,14 @@ impl BroadcastPackage { /// 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 + /// droppable package was dropped to make room and is returned — its transactions resurface /// with LDK's next periodic rebroadcast. Scheduled { dropped: Option }, /// 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. + /// The bound was reached and every waiting package is one that must not be dropped (a + /// funding or a cooperative close): the incoming package is refused and returned. Refused(BroadcastPackage), } @@ -120,8 +133,8 @@ impl RetryQueue { } /// 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`]. + /// waits or accepting it would exceed [`MAX_QUEUED_RETRIES`] with no droppable package to + /// make room with; see [`ScheduleOutcome`]. pub(crate) fn schedule( &mut self, package: BroadcastPackage, retry_at: Instant, ) -> ScheduleOutcome { @@ -137,12 +150,14 @@ impl RetryQueue { } 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 + if package.is_droppable() && self.0.len() >= MAX_QUEUED_RETRIES { + // Drop the oldest droppable package: its transactions are re-broadcast // 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()) { + // leaves its transaction confirming without a recorded candidate. Neither is a + // cooperative close, whose queued package may hold the only copy of the signed + // closing transaction. + match self.0.iter().position(|(_, _, waiting)| waiting.is_droppable()) { Some(oldest) => dropped = self.0.remove(oldest).map(|(_, _, package)| package), None => return ScheduleOutcome::Refused(package), } @@ -423,6 +438,34 @@ mod tests { BroadcastPackage::new(&[(tx, LdkTransactionType::Funding { channels: vec![] })]) } + fn test_counterparty_node_id() -> bitcoin::secp256k1::PublicKey { + use std::str::FromStr; + bitcoin::secp256k1::PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap() + } + + fn coop_close_package(tx: &Transaction) -> BroadcastPackage { + BroadcastPackage::new(&[( + tx, + LdkTransactionType::CooperativeClose { + counterparty_node_id: test_counterparty_node_id(), + channel_id: lightning::ln::types::ChannelId([13u8; 32]), + }, + )]) + } + + fn claim_package(tx: &Transaction) -> BroadcastPackage { + BroadcastPackage::new(&[( + tx, + LdkTransactionType::Claim { + counterparty_node_id: test_counterparty_node_id(), + channel_id: lightning::ln::types::ChannelId([13u8; 32]), + }, + )]) + } + fn deadline(secs: u64) -> tokio::time::Instant { tokio::time::Instant::now() + std::time::Duration::from_secs(secs) } @@ -475,10 +518,10 @@ mod tests { } /// 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 + /// the bound: the oldest droppable 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() { + async fn retry_queue_drops_the_oldest_droppable_package_at_the_bound() { fn numbered_tx(n: u32) -> Transaction { Transaction { version: bitcoin::transaction::Version::TWO, @@ -502,7 +545,7 @@ mod tests { )); } - // At the bound, an incoming non-funding package drops the oldest waiting one — not the + // At the bound, an incoming droppable 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)) { @@ -528,11 +571,11 @@ mod tests { assert!(!remaining.contains(&oldest_claim.compute_txid())); } - /// When only funding packages wait at the bound, an incoming non-funding package is refused: + /// When only funding packages wait at the bound, an incoming droppable 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() { + async fn retry_queue_refuses_a_droppable_package_over_waiting_funding_packages() { fn numbered_tx(n: u32) -> Transaction { Transaction { version: bitcoin::transaction::Version::TWO, @@ -556,4 +599,89 @@ mod tests { ScheduleOutcome::Refused(_) )); } + + /// A cooperative close is never dropped at the bound: nothing re-broadcasts it, and the + /// queued package may hold the only copy of the signed closing transaction. + #[tokio::test] + async fn retry_queue_never_drops_a_cooperative_close_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([9u8; 32]), n)], + output: vec![txout(1_000)], + } + } + + let mut retries = RetryQueue::new(); + let coop_close_tx = numbered_tx(0); + assert!(matches!( + retries.schedule(coop_close_package(&coop_close_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(claim_package(&numbered_tx(n)), deadline(2)), + ScheduleOutcome::Scheduled { dropped: None } + )); + } + + // At the bound, an incoming claim drops the oldest waiting claim — not the older + // cooperative close. + let new_claim = numbered_tx(MAX_QUEUED_RETRIES as u32); + match retries.schedule(claim_package(&new_claim), 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 cooperative close is never dropped for the bound either. + let new_coop_close_tx = numbered_tx(MAX_QUEUED_RETRIES as u32 + 1); + assert!(matches!( + retries.schedule(coop_close_package(&new_coop_close_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(&coop_close_tx.compute_txid()), + "a cooperative close is never dropped" + ); + assert!(remaining.contains(&new_coop_close_tx.compute_txid())); + assert!(!remaining.contains(&oldest_claim.compute_txid())); + } + + /// When only cooperative closes wait at the bound, an incoming claim is refused: LDK + /// re-broadcasts the claim periodically, while a dropped close would lose the only copy of + /// its signed closing transaction. + #[tokio::test] + async fn retry_queue_refuses_a_claim_over_waiting_cooperative_closes() { + 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([10u8; 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(coop_close_package(&numbered_tx(n)), deadline(2)), + ScheduleOutcome::Scheduled { dropped: None } + )); + } + + let claim = numbered_tx(MAX_QUEUED_RETRIES as u32); + assert!(matches!( + retries.schedule(claim_package(&claim), deadline(2)), + ScheduleOutcome::Refused(_) + )); + } } From a55bf730a16c2e8385880a7889e7b08ef00b62dc Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 2 Sep 2026 13:53:47 -0500 Subject: [PATCH 8/8] Fail funding payments lost to a confirmed conflict Since declining to adopt a conflicting close's confirmation, a funding payment whose transaction was double-spent stayed Pending forever -- nothing wrote a terminal status for an on-chain record -- and the sync loop kept re-queueing the dead transaction for rebroadcast on every tip change. Mark such a record Failed once a conflict from outside its candidate history has confirmed through ANTI_REORG_DELAY while neither its own transaction nor any RBF candidate can still confirm, mirroring the anti-reorg finality the Succeeded transition already assumes. Removing the payment's pending entry then stops the re-queueing. Settling also removes the entry that maps candidate txids to the record, so a later wallet event for a dead candidate falls back to keying by that candidate's txid -- which, for the first candidate, is the record's own id. Skip such events rather than let the generic handling resurrect the settled record, and let a replayed replacement event finish an entry removal a crash interrupted instead of stamping the terminal status into the leftover entry. Implemented with Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 684 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 682 insertions(+), 2 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index c8477c9d7..a04ace2f5 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -12,6 +12,7 @@ use std::str::FromStr; use std::sync::{Arc, Mutex}; use bdk_chain::spk_client::{FullScanRequest, SyncRequest}; +use bdk_chain::ChainPosition; use bdk_wallet::descriptor::ExtendedDescriptor; use bdk_wallet::error::{BuildFeeBumpError, CreateTxError}; #[allow(deprecated)] @@ -369,6 +370,17 @@ impl Wallet { }, } + // The fallback id belongs to a settled funding payment whose entry is gone: + // skip rather than resurrect it (see `has_funding_record`). + if self.has_funding_record(&payment_id).await? { + log_debug!( + self.logger, + "Skipping wallet event for transaction {} of a settled funding payment", + txid, + ); + continue; + } + let payment = { let locked_wallet = self.inner.lock().expect("lock"); self.create_payment_from_tx( @@ -455,8 +467,16 @@ impl Wallet { txid, status: ConfirmationStatus::Unconfirmed, .. - } if payment.details.direction == PaymentDirection::Outbound => { - unconfirmed_outbound_txids.push(txid); + } => { + if self + .fail_funding_payment_lost_to_conflict(&payment, new_tip.height) + .await? + { + continue; + } + if payment.details.direction == PaymentDirection::Outbound { + unconfirmed_outbound_txids.push(txid); + } }, _ => {}, } @@ -516,6 +536,17 @@ impl Wallet { }, } + // The fallback id belongs to a settled funding payment whose entry is gone: + // skip rather than resurrect it (see `has_funding_record`). + if self.has_funding_record(&payment_id).await? { + log_debug!( + self.logger, + "Skipping wallet event for transaction {} of a settled funding payment", + txid, + ); + continue; + } + let payment = { let locked_wallet = self.inner.lock().expect("lock"); self.create_payment_from_tx( @@ -565,6 +596,18 @@ impl Wallet { payment_id, ); let payment = stored_payment.ok_or(Error::InvalidPaymentId)?; + + // A terminal record means the entry is the leftover of an interrupted settle + // — the record write landed, the entry removal was lost to a crash — and this + // event is the restart's replay of the same transition. Re-embedding the + // record would stamp the terminal status into the entry and hide it from the + // pending listing that repairs such leftovers; finish the interrupted removal + // instead. + if payment.status != PaymentStatus::Pending { + self.pending_payment_store.remove(&payment_id).await?; + continue; + } + let pending_payment_details = self.create_pending_payment_from_tx(payment, conflict_txids.clone()); @@ -598,6 +641,17 @@ impl Wallet { }, } + // The fallback id belongs to a settled funding payment whose entry is gone: + // skip rather than resurrect it (see `has_funding_record`). + if self.has_funding_record(&payment_id).await? { + log_debug!( + self.logger, + "Skipping wallet event for transaction {} of a settled funding payment", + txid, + ); + continue; + } + let payment = { let locked_wallet = self.inner.lock().expect("lock"); self.create_payment_from_tx( @@ -623,6 +677,152 @@ impl Wallet { Ok(()) } + /// Whether a funding-classified record exists under the given id. A funding record's id is + /// anchored to its first candidate's txid, so a wallet event for that transaction falls back + /// to this id whenever the pending entry no longer maps it — which only happens once the + /// negotiation settled and the entry was removed. The generic event handling must then skip + /// its write: merging a wallet-view `Pending` payment into the settled record would resurrect + /// it with figures no classification derived. + async fn has_funding_record(&self, payment_id: &PaymentId) -> Result { + Ok(self.payment_store.get(payment_id).await?.is_some_and(|payment| { + matches!( + payment.kind, + PaymentKind::Onchain { + tx_type: Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. } + ), + .. + } + ) + })) + } + + /// Fails a funding payment whose transaction has irrevocably lost a conflict: a transaction + /// outside the record's candidate history — e.g. a channel close double-spending a pending + /// splice's shared input — has confirmed through [`ANTI_REORG_DELAY`] while neither the + /// record's transaction nor any candidate is canonical anymore. Returns whether the payment + /// was failed; failing also removes the pending entry, dropping the dead record from the + /// tip-change pass. (Its transaction was already excluded from rebroadcast by the same + /// canonical-only `get_tx` gate used below.) + /// + /// Only funding-classified records are considered: nothing re-submits a replaced funding + /// transaction under the same record (an RBF round is a new candidate), so a buried foreign + /// conflict is final for them. The liveness check guards the case where the conflict + /// double-spent only one round of the negotiation: as long as some candidate — including one + /// classification hasn't recorded yet — can still confirm, the record must stay pending. + async fn fail_funding_payment_lost_to_conflict( + &self, payment: &PendingPaymentDetails, tip_height: u32, + ) -> Result { + match payment.details.kind { + PaymentKind::Onchain { + status: ConfirmationStatus::Unconfirmed, + tx_type: + Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. }, + ), + .. + } => {}, + _ => return Ok(false), + } + if payment.conflicting_txids.is_empty() { + return Ok(false); + } + + // Serialize with classification, whose retries extend the candidate history: the + // decision below must see that history in its settled form, and holding the lock keeps a + // concurrent write from resurrecting the entry removed at the end. + let _guard = self.funding_payment_update_lock.lock().await; + + // Re-read the entry under the lock; the listing snapshot may predate a classification. + let entry = match self.pending_payment_store.get(&payment.details.id).await? { + Some(entry) => entry, + None => return Ok(false), + }; + let record_txid = match entry.details.kind { + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: + Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. }, + ), + } => txid, + _ => return Ok(false), + }; + + let foreign_conflicts: Vec = entry + .conflicting_txids + .iter() + .copied() + .filter(|conflict| *conflict != record_txid && entry.candidate(*conflict).is_none()) + .collect(); + if foreign_conflicts.is_empty() { + return Ok(false); + } + + let lost = { + let locked_wallet = self.inner.lock().expect("lock"); + // `get_tx` is canonical-only: a transaction that lost to a confirmed conflict + // returns `None`, while one that can still confirm is `Some`. + let a_candidate_is_live = locked_wallet.get_tx(record_txid).is_some() + || entry.candidates.iter().any(|c| locked_wallet.get_tx(c.txid).is_some()); + !a_candidate_is_live + && foreign_conflicts.iter().any(|conflict| { + match locked_wallet.get_tx(*conflict).map(|tx| tx.chain_position) { + Some(ChainPosition::Confirmed { anchor, .. }) => { + tip_height >= anchor.block_id.height + ANTI_REORG_DELAY - 1 + }, + _ => false, + } + }) + }; + if !lost { + return Ok(false); + } + + // As with graduation, decide from the live record and write only the status. A record + // already `Failed` — a prior pass whose entry removal below was lost to a crash — still + // matches, no-ops the update, and gets its lingering entry removed. + let payment_id = entry.details.id; + let mut failed = false; + self.payment_store + .mutate(&payment_id, |existing| { + let current = existing?; + match current.kind { + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: + Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. }, + ), + } if txid == record_txid => { + failed = true; + let mut update = PaymentDetailsUpdate::new(payment_id); + update.status = Some(PaymentStatus::Failed); + let mut updated = current.clone(); + updated.update(update).then_some(updated) + }, + _ => None, + } + }) + .await?; + if failed { + self.pending_payment_store.remove(&payment_id).await?; + log_info!( + self.logger, + "Failed funding payment {}: transaction {} lost to a conflicting transaction confirmed beyond the reorg depth", + payment_id, + record_txid, + ); + } + Ok(failed) + } + #[allow(deprecated)] pub(crate) async fn create_funding_transaction( &self, output_script: ScriptBuf, amount: Amount, confirmation_target: ConfirmationTarget, @@ -3799,6 +3999,57 @@ mod tests { } } + /// Inserts `tx` into the BDK wallet as canonically confirmed at `height`, extending the + /// local chain to that height. + fn insert_confirmed_tx(wallet: &Wallet, tx: Transaction, height: u32) { + let txid = tx.compute_txid(); + let mut locked = wallet.inner.lock().unwrap(); + let block = + BlockId { height, hash: bitcoin::BlockHash::from_byte_array([height as u8; 32]) }; + let chain = locked.latest_checkpoint().insert(block); + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.txs = vec![Arc::new(tx)]; + tx_update.anchors = + [(ConfirmationBlockTime { block_id: block, confirmation_time: 100 }, txid)].into(); + locked + .apply_update(Update { tx_update, chain: Some(chain), ..Default::default() }) + .unwrap(); + } + + /// Inserts `tx` into the BDK wallet as canonically unconfirmed (seen in the mempool). + fn insert_unconfirmed_tx(wallet: &Wallet, tx: Transaction) { + let txid = tx.compute_txid(); + let mut locked = wallet.inner.lock().unwrap(); + let mut tx_update = bdk_chain::TxUpdate::default(); + tx_update.txs = vec![Arc::new(tx)]; + tx_update.seen_ats = [(txid, 100)].into(); + locked.apply_update(Update { tx_update, ..Default::default() }).unwrap(); + } + + /// Builds a transaction paying a wallet address, spending an outpoint derived from + /// `input_byte` (distinct bytes yield non-conflicting transactions). + fn wallet_paying_tx(wallet: &Wallet, input_byte: u8) -> Transaction { + let script_pubkey = wallet + .inner + .lock() + .unwrap() + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(); + Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: vec![bitcoin::TxIn { + previous_output: OutPoint { + txid: Txid::from_byte_array([input_byte; 32]), + vout: 0, + }, + ..Default::default() + }], + output: vec![TxOut { value: Amount::from_sat(90_000), script_pubkey }], + } + } + #[test] fn funding_reclassification_update_substitutes_the_confirmed_candidate() { let confirmed_txid = Txid::from_byte_array([1u8; 32]); @@ -4200,6 +4451,435 @@ mod tests { } } + /// Continues the story above: once the conflicting close confirms through the anti-reorg + /// depth, the splice's funding transaction can never confirm — its shared input is spent for + /// good. The record must fail rather than stay `Pending` forever, and removing the pending + /// entry stops the dead transaction's rebroadcast on every tip change. + #[tokio::test] + async fn funding_payment_fails_once_a_foreign_conflict_confirms_to_depth() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let close_tx = wallet_paying_tx(&wallet, 3); + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![close_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + // The close is canonically confirmed; the splice transaction, having lost the conflict, + // is no longer canonical (here: never inserted at all). + insert_confirmed_tx(&wallet, close_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed); + match &payment.kind { + PaymentKind::Onchain { txid, status, tx_type } => { + assert_eq!(*txid, splice_txid, "failing must not adopt the conflict's txid"); + assert!(matches!(status, ConfirmationStatus::Unconfirmed)); + assert!(matches!(tx_type, Some(TransactionType::InteractiveFunding { .. }))); + }, + kind => panic!("unexpected kind {:?}", kind), + } + assert_eq!(payment.amount_msat, Some(1_000_000)); + assert_eq!(payment.fee_paid_msat, Some(500)); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none(), + "the entry must go so the dead transaction stops being rebroadcast" + ); + } + + /// A confirmed conflict that is one of the record's own candidates is RBF resolution, not a + /// loss: classification adopts it into the record, so the failure pass must leave the record + /// alone. + #[tokio::test] + async fn funding_payment_survives_a_confirmed_conflict_that_is_a_candidate() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let bumped_tx = wallet_paying_tx(&wallet, 3); + let bumped_txid = bumped_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![ + FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + FundingTxCandidate { + txid: bumped_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(600), + }, + ]; + let details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![bumped_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + insert_confirmed_tx(&wallet, bumped_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Pending); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_some(), + "the entry must survive for classification to adopt the confirmed candidate" + ); + } + + /// A foreign conflict that has confirmed but not yet through the anti-reorg depth may still + /// be reorged out, letting the funding transaction confirm after all; the record must stay + /// pending until the conflict's confirmation is final. + #[tokio::test] + async fn funding_payment_survives_a_foreign_conflict_short_of_depth() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let close_tx = wallet_paying_tx(&wallet, 3); + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![close_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + insert_confirmed_tx(&wallet, close_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 2), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Pending); + assert!(wallet.pending_payment_store.get(&payment_id).await.unwrap().is_some()); + } + + /// A conflict may double-spend only one round of the negotiation — e.g. it shares an input + /// with an RBF attempt but not with the original candidate. While any candidate is still + /// canonical it can still confirm, so the record must stay pending. + #[tokio::test] + async fn funding_payment_survives_while_a_candidate_can_still_confirm() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let conflict_tx = wallet_paying_tx(&wallet, 3); + let conflict_txid = conflict_tx.compute_txid(); + // A live candidate: spends a different outpoint, so the conflict didn't kill it. + let live_candidate_tx = wallet_paying_tx(&wallet, 4); + let live_candidate_txid = live_candidate_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![ + FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + FundingTxCandidate { + txid: live_candidate_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(600), + }, + ]; + let details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![conflict_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + insert_confirmed_tx(&wallet, conflict_tx, 5); + insert_unconfirmed_tx(&wallet, live_candidate_tx); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Pending); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_some(), + "a candidate can still confirm, so the record must stay pending" + ); + } + + /// The failure write pair is record first, entry second: a crash in between leaves a + /// `Failed` record with a lingering entry. The next tip pass must finish the job — remove + /// the entry without disturbing the record. + #[tokio::test] + async fn a_failed_funding_payment_with_a_lingering_entry_is_cleaned_up() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let close_tx = wallet_paying_tx(&wallet, 3); + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let mut recorded = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + recorded.status = PaymentStatus::Failed; + recorded.latest_update_timestamp = 7; + wallet.payment_store.insert_or_update(recorded).await.unwrap(); + + // The entry embeds the pre-failure snapshot, as a crash between the two writes leaves it. + let snapshot = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let entry = PendingPaymentDetails::new(snapshot, vec![close_txid], candidates); + wallet.pending_payment_store.insert_or_update(entry).await.unwrap(); + + insert_confirmed_tx(&wallet, close_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed); + assert_eq!(payment.latest_update_timestamp, 7, "the repair pass must not rewrite"); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none(), + "the lingering entry must be removed" + ); + } + + /// A crash between the failure's record write and its entry removal loses the wallet + /// changeset too, so the restart's catch-up sync replays the same events: `TxReplaced` for + /// the dead funding transaction resolves through the lingering entry to the already-`Failed` + /// record. Re-embedding that record would stamp `Failed` into the entry and hide it from the + /// pending listing that repairs it; the replay must instead finish the interrupted removal. + #[tokio::test] + async fn replayed_replacement_finishes_an_interrupted_failure() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let close_tx = wallet_paying_tx(&wallet, 3); + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let mut recorded = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + recorded.status = PaymentStatus::Failed; + recorded.latest_update_timestamp = 7; + wallet.payment_store.insert_or_update(recorded).await.unwrap(); + + let snapshot = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let entry = PendingPaymentDetails::new(snapshot, vec![close_txid], candidates); + wallet.pending_payment_store.insert_or_update(entry).await.unwrap(); + + insert_confirmed_tx(&wallet, close_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let events = vec![ + WalletEvent::TxReplaced { + txid: splice_txid, + tx: Arc::new(dummy_tx()), + conflicts: vec![(0, close_txid)], + }, + WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }, + ]; + wallet.update_payment_store(events).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed); + assert_eq!(payment.latest_update_timestamp, 7, "the replay must not rewrite the record"); + assert!( + wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none(), + "the replay must finish the interrupted entry removal" + ); + } + + /// A funding record's id is anchored to its first candidate's txid. Once the payment settles + /// and its entry is removed, a wallet event for that candidate no longer resolves through the + /// candidate history — the fallback keys it by its own txid, colliding with the record's id. + /// Recording the event there would merge a fresh wallet-view `Pending` payment into the + /// terminal record; such events must be skipped. + #[tokio::test] + async fn candidate_event_does_not_resurrect_a_settled_funding_payment() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + // The record's id derives from the first candidate r1; its txid rotated to the RBF round + // r2. The payment failed and its pending entry is gone. + let r1 = Txid::from_byte_array([2u8; 32]); + let r2 = Txid::from_byte_array([4u8; 32]); + let payment_id = PaymentId(r1.to_byte_array()); + let mut recorded = interactive_funding_details(payment_id, r2, Some(1_000_000), Some(600)); + recorded.status = PaymentStatus::Failed; + recorded.latest_update_timestamp = 7; + wallet.payment_store.insert_or_update(recorded).await.unwrap(); + + // r1 reappears in the mempool after the failure... + let event = + WalletEvent::TxUnconfirmed { txid: r1, tx: Arc::new(dummy_tx()), old_block_time: None }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed, "the record must not resurrect"); + assert!(matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == r2)); + assert_eq!(payment.latest_update_timestamp, 7); + assert!(wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none()); + + // ...and even confirms: the record settled as `Failed` and must stay that way. + let event = WalletEvent::TxConfirmed { + txid: r1, + tx: Arc::new(dummy_tx()), + block_time: confirmed_block_time(5), + old_block_time: None, + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed, "the record must not resurrect"); + assert!(matches!(payment.kind, PaymentKind::Onchain { txid, .. } if txid == r2)); + assert_eq!(payment.latest_update_timestamp, 7); + assert!(wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none()); + } + + /// The failure transition must apply regardless of the payment's direction: a splice-out + /// records as `Inbound` (funds return to the wallet) and dies to a conflicting close the + /// same way an outbound one does. + #[tokio::test] + async fn inbound_funding_payment_fails_once_a_foreign_conflict_confirms_to_depth() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let close_tx = wallet_paying_tx(&wallet, 3); + let close_txid = close_tx.compute_txid(); + + let splice_txid = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![FundingTxCandidate { + txid: splice_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let mut details = + interactive_funding_details(payment_id, splice_txid, Some(1_000_000), Some(500)); + details.direction = PaymentDirection::Inbound; + wallet.persist_funding_payment(details, candidates).await.unwrap(); + wallet + .pending_payment_store + .update(PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: Some(vec![close_txid]), + candidates: Vec::new(), + }) + .await + .unwrap(); + + insert_confirmed_tx(&wallet, close_tx, 5); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { + old_tip: block_id(9), + new_tip: block_id(5 + ANTI_REORG_DELAY - 1), + }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).await.unwrap().unwrap(); + assert_eq!(payment.status, PaymentStatus::Failed); + assert!(wallet.pending_payment_store.get(&payment_id).await.unwrap().is_none()); + } + /// A funding-typed broadcast that doesn't touch the on-chain wallet must not be recorded. /// LDK re-broadcasts a promoted-but-unconfirmed 0conf splice through its generic funding /// path, so a splice the interactive-funding classification deliberately declined — no local