From 98b11a5856593b778de73d585b6d65c418291365 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 1 Jul 2026 18:49:09 -0500 Subject: [PATCH 01/11] Preserve funding-payment confirmation state on late reclassification Funding broadcasts are classified into payment records off the broadcaster's queue, which can run after wallet sync has already recorded the transaction -- for instance when the counterparty's broadcast of the funding transaction is observed by wallet sync first. In that case the classification overwrote a record wallet sync had already advanced, downgrading a confirmed or graduated funding payment back to unconfirmed/pending. Merge only the classification and our contribution figures into an existing record, leaving the confirmation state that the wallet-sync events own in place. Raised by Codex in the review of #888. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/payment/store.rs | 115 +++++++++++++++++++++++++++++++++++++++++++ src/wallet/mod.rs | 28 +++++++++-- 2 files changed, 139 insertions(+), 4 deletions(-) diff --git a/src/payment/store.rs b/src/payment/store.rs index d2b92747a2..df2e9dc099 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -712,6 +712,33 @@ impl PaymentDetailsUpdate { tx_type: None, } } + + /// Builds an update that merges a freshly-classified funding payment's classification + /// (`tx_type`), broadcast txid, and our contribution figures (amount/fee) into an existing + /// record, while leaving the top-level [`PaymentStatus`] and the on-chain + /// [`ConfirmationStatus`] untouched. + /// + /// Funding classification runs off the broadcaster queue and can land *after* wallet sync has + /// already advanced a record's confirmation state (e.g. when LDK re-broadcasts a still-pending + /// funding transaction on restart, or when the counterparty's broadcast is observed first). + /// Merging only the funding-specific fields keeps such a late classification from downgrading a + /// `Confirmed`/`Succeeded` payment back to `Unconfirmed`/`Pending`; the confirmation state is + /// owned by the wallet-sync events instead. + /// + /// The txid and figures are taken from the freshly broadcast (active) candidate. LDK only + /// re-broadcasts the active/confirmed funding candidate, so for an already-confirmed record + /// these equal what graduation stamped and the overwrite is a no-op; we rely on that invariant + /// rather than gating the txid/amount/fee merge on the stored confirmation state. + pub(crate) fn funding_reclassification(details: PaymentDetails) -> Self { + let mut update = Self::new(details.id); + update.amount_msat = Some(details.amount_msat); + update.fee_paid_msat = Some(details.fee_paid_msat); + if let PaymentKind::Onchain { txid, tx_type, .. } = details.kind { + update.txid = Some(txid); + update.tx_type = Some(tx_type); + } + update + } } impl From<&PaymentDetails> for PaymentDetailsUpdate { @@ -1022,6 +1049,94 @@ mod tests { } } + #[test] + fn funding_reclassification_does_not_downgrade_an_advanced_record() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // A splice funding payment wallet sync has already advanced to Succeeded/Confirmed. + let txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(txid.to_byte_array()); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let advanced = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: tx_type.clone(), + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Succeeded, + ); + + // A fresh funding classification for the same payment is always Pending/Unconfirmed. + let fresh = PaymentDetails::new( + id, + PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The naive full update `insert_or_update` applied before the fix downgrades both the + // top-level status and the on-chain confirmation status — the bug Codex flagged. + let mut downgraded = advanced.clone(); + downgraded.update((&fresh).into()); + assert_eq!( + downgraded.status, + PaymentStatus::Pending, + "a full update from a fresh classification downgrades the top-level status", + ); + assert!( + matches!( + downgraded.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + ), + "a full update from a fresh classification downgrades the confirmation status", + ); + + // The narrowed reclassification update merges only the funding fields and preserves the + // advanced confirmation state that wallet sync owns. + let mut merged = advanced.clone(); + merged.update(PaymentDetailsUpdate::funding_reclassification(fresh)); + assert_eq!( + merged.status, + PaymentStatus::Succeeded, + "reclassification must not downgrade the top-level status", + ); + assert!( + matches!( + merged.kind, + PaymentKind::Onchain { + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } + ), + "reclassification must preserve the confirmation status and keep the funding tx_type", + ); + // The contribution-derived figures from the fresh classification ARE merged in, replacing + // the existing record's: they are authoritative (the wallet can't recompute our share of a + // shared funding output), so the merge must carry them. + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); + } + #[derive(Clone, Debug, PartialEq, Eq)] struct LegacyBolt11JitKind { hash: PaymentHash, diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8d9d521eb..5b53ef292f 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -56,7 +56,8 @@ use persist::KVStoreWalletPersister; use crate::config::Config; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; -use crate::payment::store::ConfirmationStatus; +use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; +use crate::payment::store::{ConfirmationStatus, PaymentDetailsUpdate}; use crate::payment::{ FundingTxCandidate, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, PendingPaymentDetails, TransactionType, @@ -1398,9 +1399,28 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { - self.payment_store.insert_or_update(details.clone()).await?; - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.insert_or_update(pending).await?; + if !self.payment_store.contains_key(&details.id) { + // First time we record this funding payment: store it and index it for graduation. + self.payment_store.insert_or_update(details.clone()).await?; + let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); + self.pending_payment_store.insert_or_update(pending).await?; + } else { + // An earlier candidate or a racing wallet sync already recorded this payment. Merge only + // the classification (`tx_type`) and our contribution figures, which the wallet can't + // recompute; the confirmation state is owned by wallet-sync events, so a late + // classification must not move it (which would downgrade an already-Confirmed/Succeeded + // record). `update` is a no-op when the entry is absent, so the pending index is not + // re-created for a payment the graduation path already removed. + let update = PaymentDetailsUpdate::funding_reclassification(details); + let pending_update = PendingPaymentDetailsUpdate { + id: update.id, + payment_update: Some(update.clone()), + conflicting_txids: None, + candidates, + }; + self.payment_store.update(update).await?; + self.pending_payment_store.update(pending_update).await?; + } Ok(()) } From cb99564c8065a92cf44ffb9f208650ba1c8b125f Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 28 Jul 2026 18:24:00 -0500 Subject: [PATCH 02/11] f - Make funding-payment classification write atomic with its existence check Co-Authored-By: Claude --- src/data_store.rs | 149 +++++++++++++++++++++++++++ src/payment/pending_payment_store.rs | 74 +++++++++++++ src/payment/store.rs | 105 +++++++++++++++++++ src/wallet/mod.rs | 55 ++++++---- 4 files changed, 362 insertions(+), 21 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index b1ed816df9..b7748b172e 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -40,6 +40,13 @@ pub(crate) enum DataStoreUpdateResult { NotFound, } +#[derive(PartialEq, Eq, Debug, Clone, Copy)] +pub(crate) enum DataStoreUpdateOrInsertResult { + Inserted, + Updated, + Unchanged, +} + pub(crate) struct DataStore where L::Target: LdkLogger, @@ -81,6 +88,11 @@ where Ok(updated) } + /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the + /// object's full update ([`StorableObject::to_update`]) into it instead of replacing it. + /// + /// Unlike [`Self::update_or_insert`], the caller does not choose what is merged into an + /// existing entry: the full update is always applied. pub(crate) async fn insert_or_update(&self, object: SO) -> Result { let _guard = self.mutation_lock.lock().await; @@ -170,6 +182,47 @@ where Ok(DataStoreUpdateResult::Updated) } + /// Applies `update` when an object with its id already exists, or inserts `object` when none + /// does. + /// + /// Like [`Self::update`], but falls back to inserting `object` instead of returning + /// [`DataStoreUpdateResult::NotFound`]. Unlike [`Self::insert_or_update`], the caller chooses + /// exactly what is merged into an existing entry: `update` may carry less than the full + /// object. + /// + /// The existence check and the write share one critical section of the mutation lock, so a + /// concurrent writer cannot land in between and later have its state clobbered by the insert + /// fallback — the check-then-act race that separate [`Self::contains_key`] + + /// [`Self::insert_or_update`] calls reintroduce. + pub(crate) async fn update_or_insert( + &self, update: SO::Update, object: SO, + ) -> Result { + debug_assert!(update.id() == object.id(), "update and object must share an id"); + let _guard = self.mutation_lock.lock().await; + + let id = update.id(); + let (data_to_persist, result) = { + let locked_objects = self.objects.lock().expect("lock"); + match locked_objects.get(&id) { + Some(existing_object) => { + let mut updated_object = existing_object.clone(); + if updated_object.update(update) { + (Some(updated_object), DataStoreUpdateOrInsertResult::Updated) + } else { + (None, DataStoreUpdateOrInsertResult::Unchanged) + } + }, + None => (Some(object), DataStoreUpdateOrInsertResult::Inserted), + } + }; + + if let Some(object) = data_to_persist { + self.persist(&object).await?; + self.objects.lock().expect("lock").insert(id, object); + } + Ok(result) + } + /// Returns in-memory objects matching `f`. /// /// The async mutation lock serializes writers, but this synchronous reader cannot wait on it. @@ -403,6 +456,102 @@ mod tests { assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await); } + #[tokio::test] + async fn update_or_insert_inserts_when_absent() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let primary_namespace = "datastore_test_primary".to_string(); + let secondary_namespace = "datastore_test_secondary".to_string(); + let data_store: DataStore> = DataStore::new( + Vec::new(), + primary_namespace.clone(), + secondary_namespace.clone(), + Arc::clone(&store), + logger, + ); + + let id = TestObjectId { id: [42u8; 4] }; + let object = TestObject { id, data: [23u8; 3] }; + let update = TestObjectUpdate { id, data: [25u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Inserted), + data_store.update_or_insert(update, object).await + ); + + // The insert path stores the fallback object as-is; the update is not applied to it. + assert_eq!(Some(object), data_store.get(&id)); + let store_key = id.encode_to_hex_str(); + assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key) + .await + .is_ok()); + } + + #[tokio::test] + async fn update_or_insert_applies_update_when_present() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store: DataStore> = DataStore::new( + vec![existing_object], + "datastore_test_primary".to_string(), + "datastore_test_secondary".to_string(), + store, + logger, + ); + + // When an entry exists, only the update is applied; the fallback object must not replace + // it. + let update = TestObjectUpdate { id, data: [24u8; 3] }; + let object = TestObject { id, data: [99u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Updated), + data_store.update_or_insert(update, object).await + ); + assert_eq!(data_store.get(&id).unwrap().data, [24u8; 3]); + } + + #[tokio::test] + async fn update_or_insert_returns_unchanged_without_persisting() { + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + // A no-op update returns `Unchanged` without attempting to persist (the store fails all + // writes) and without falling back to the object. + let update = TestObjectUpdate { id, data: [23u8; 3] }; + let object = TestObject { id, data: [99u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Unchanged), + data_store.update_or_insert(update, object).await + ); + assert_eq!(Some(existing_object), data_store.get(&id)); + } + + #[tokio::test] + async fn update_or_insert_does_not_mutate_memory_if_persist_fails() { + let existing_id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id: existing_id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + let update = TestObjectUpdate { id: existing_id, data: [24u8; 3] }; + let object = TestObject { id: existing_id, data: [24u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.update_or_insert(update, object).await + ); + assert_eq!(Some(existing_object), data_store.get(&existing_id)); + + let new_id = TestObjectId { id: [55u8; 4] }; + let new_object = TestObject { id: new_id, data: [34u8; 3] }; + let new_update = TestObjectUpdate { id: new_id, data: [34u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.update_or_insert(new_update, new_object).await + ); + assert!(data_store.get(&new_id).is_none()); + } + #[tokio::test] async fn insert_or_update_does_not_mutate_memory_if_persist_fails() { let existing_id = TestObjectId { id: [42u8; 4] }; diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index f5f2fa40a2..b822427dfb 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -242,4 +242,78 @@ mod tests { "current txid must not remain in its own conflict list" ); } + + #[test] + fn funding_classification_pending_update_preserves_mirrored_confirmation() { + use bitcoin::BlockHash; + + use crate::payment::store::PaymentDetailsUpdate; + + let txid = test_txid(7); + let payment_id = PaymentId(txid.to_byte_array()); + + // A pending entry wallet sync has already mirrored a confirmation into (via + // `apply_funding_status_update`) while classification was still writing its two stores. + let confirmed_details = PaymentDetails::new( + payment_id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + let mirrored = PendingPaymentDetails::new(confirmed_details, Vec::new(), Vec::new()); + + // A fresh classification is always Unconfirmed and carries the candidate history; its + // figures are the active candidate's. + let fresh = pending_onchain_payment(payment_id, txid); + let candidates = vec![FundingTxCandidate { + txid, + amount_msat: fresh.amount_msat, + fee_paid_msat: fresh.fee_paid_msat, + }]; + + // The old fresh-insert path merged the full fresh record, downgrading the mirrored + // confirmation. + let mut downgraded = mirrored.clone(); + let full_update = + PendingPaymentDetails::new(fresh.clone(), Vec::new(), candidates.clone()).to_update(); + assert!(downgraded.update(full_update)); + assert!( + matches!( + downgraded.details.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + ), + "a full merge of a fresh classification downgrades a mirrored confirmation", + ); + + // The narrow classification update merges the figures and candidates while preserving the + // confirmation state wallet sync owns. + let mut merged = mirrored.clone(); + let narrow_update = PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)), + conflicting_txids: None, + candidates: candidates.clone(), + }; + assert!(merged.update(narrow_update)); + assert!( + matches!( + merged.details.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + ), + "a narrow classification update must not downgrade a mirrored confirmation", + ); + assert_eq!(merged.candidates, candidates); + assert_eq!(merged.details.amount_msat, Some(1_000)); + assert_eq!(merged.details.fee_paid_msat, Some(100)); + } } diff --git a/src/payment/store.rs b/src/payment/store.rs index df2e9dc099..17819b03a6 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -1137,6 +1137,111 @@ mod tests { assert_eq!(merged.fee_paid_msat, Some(500)); } + #[tokio::test] + async fn funding_classification_update_or_insert_preserves_advanced_record() { + use bitcoin::hashes::Hash; + use lightning::util::test_utils::TestLogger; + use std::str::FromStr; + use std::sync::Arc; + + use crate::data_store::{DataStore, DataStoreUpdateOrInsertResult}; + use crate::io::test_utils::InMemoryStore; + use crate::types::{DynStore, DynStoreWrapper}; + + let txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(txid.to_byte_array()); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + // A funding payment wallet sync has already advanced to Succeeded/Confirmed. + let advanced = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: tx_type.clone(), + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Succeeded, + ); + // A fresh funding classification for the same payment is always Pending/Unconfirmed. + let fresh = PaymentDetails::new( + id, + PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + let new_store = |seed: Vec| { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + DataStore::>::new( + seed, + "payment_test_primary".to_string(), + "payment_test_secondary".to_string(), + store, + logger, + ) + }; + + // The pre-fix fresh-insert path — a full `insert_or_update` merge landing after a racing + // wallet sync already advanced the record — downgrades it. + let store = new_store(vec![advanced.clone()]); + store.insert_or_update(fresh.clone()).await.unwrap(); + let downgraded = store.get(&id).unwrap(); + assert_eq!( + downgraded.status, + PaymentStatus::Pending, + "a full merge of a fresh classification downgrades an advanced record", + ); + + // `update_or_insert` applies only the narrow reclassification when a record exists — no + // matter when it appeared — preserving the confirmation state wallet sync owns while + // still merging the contribution-derived figures. + let store = new_store(vec![advanced.clone()]); + let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Updated), + store.update_or_insert(update, fresh.clone()).await + ); + let merged = store.get(&id).unwrap(); + assert_eq!(merged.status, PaymentStatus::Succeeded); + assert!(matches!( + merged.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + )); + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); + + // And it inserts the fresh details when no record exists yet. + let store = new_store(Vec::new()); + let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Inserted), + store.update_or_insert(update, fresh).await + ); + let inserted = store.get(&id).unwrap(); + assert_eq!(inserted.status, PaymentStatus::Pending); + assert!(matches!( + inserted.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + )); + } + #[derive(Clone, Debug, PartialEq, Eq)] struct LegacyBolt11JitKind { hash: PaymentHash, diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 5b53ef292f..2e6c41a436 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; +use crate::data_store::DataStoreUpdateOrInsertResult; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -1399,27 +1400,39 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { - if !self.payment_store.contains_key(&details.id) { - // First time we record this funding payment: store it and index it for graduation. - self.payment_store.insert_or_update(details.clone()).await?; - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.insert_or_update(pending).await?; - } else { - // An earlier candidate or a racing wallet sync already recorded this payment. Merge only - // the classification (`tx_type`) and our contribution figures, which the wallet can't - // recompute; the confirmation state is owned by wallet-sync events, so a late - // classification must not move it (which would downgrade an already-Confirmed/Succeeded - // record). `update` is a no-op when the entry is absent, so the pending index is not - // re-created for a payment the graduation path already removed. - let update = PaymentDetailsUpdate::funding_reclassification(details); - let pending_update = PendingPaymentDetailsUpdate { - id: update.id, - payment_update: Some(update.clone()), - conflicting_txids: None, - candidates, - }; - self.payment_store.update(update).await?; - self.pending_payment_store.update(pending_update).await?; + // Deciding between a fresh insert and a merge must be atomic with the write: a racing + // wallet sync can record and advance this payment between a separate existence check and + // the write, and a full merge of the fresh Pending/Unconfirmed details would then + // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds + // the store's mutation lock across the whole decision: when a record exists — no matter + // when it appeared — only the classification (`tx_type`), the broadcast txid, and our + // contribution figures are merged, which the wallet can't recompute; otherwise the fresh + // details are inserted. + let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); + let pending_update = PendingPaymentDetailsUpdate { + id: update.id, + payment_update: Some(update.clone()), + conflicting_txids: None, + candidates: candidates.clone(), + }; + match self.payment_store.update_or_insert(update, details.clone()).await? { + DataStoreUpdateOrInsertResult::Inserted => { + // First time we record this funding payment: index it for graduation. Wallet sync + // can still land between the payment-store write above and this one and mirror an + // advanced confirmation into the pending store, so this write makes the same + // atomic decision: merge narrowly into an entry that appeared, insert the fresh + // one otherwise. + let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); + self.pending_payment_store.update_or_insert(pending_update, pending).await?; + }, + DataStoreUpdateOrInsertResult::Updated | DataStoreUpdateOrInsertResult::Unchanged => { + // An earlier candidate or a racing wallet sync already recorded this payment. + // `update` is a no-op when the pending entry is absent, so the index is not + // re-created for a payment the graduation path already removed. (A graduated + // payment always has a payment-store record, so it cannot take the `Inserted` + // branch above and be re-indexed.) + self.pending_payment_store.update(pending_update).await?; + }, } Ok(()) } From 0e44736e3a0cc8be35a73b83d2ca6ff8098920f7 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 28 Jul 2026 18:45:05 -0500 Subject: [PATCH 03/11] f - Keep a confirmed funding record's txid and figures on late classification Co-Authored-By: Claude --- src/payment/pending_payment_store.rs | 8 +- src/payment/store.rs | 155 ++++++++++++++++++++++----- src/wallet/mod.rs | 6 +- 3 files changed, 135 insertions(+), 34 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index b822427dfb..2257de43cb 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -295,8 +295,8 @@ mod tests { "a full merge of a fresh classification downgrades a mirrored confirmation", ); - // The narrow classification update merges the figures and candidates while preserving the - // confirmation state wallet sync owns. + // The narrow classification update merges the candidates while preserving the + // confirmation state wallet sync owns and the confirmed candidate's figures. let mut merged = mirrored.clone(); let narrow_update = PendingPaymentDetailsUpdate { id: payment_id, @@ -313,7 +313,7 @@ mod tests { "a narrow classification update must not downgrade a mirrored confirmation", ); assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(1_000)); - assert_eq!(merged.details.fee_paid_msat, Some(100)); + assert_eq!(merged.details.amount_msat, Some(2_000_000)); + assert_eq!(merged.details.fee_paid_msat, Some(999)); } } diff --git a/src/payment/store.rs b/src/payment/store.rs index 17819b03a6..00fcffcff8 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -243,12 +243,24 @@ impl StorableObject for PaymentDetails { } } - if let Some(amount_opt) = update.amount_msat { - update_if_necessary!(self.amount_msat, amount_opt); - } + // Once an on-chain record is confirmed, its txid and figures describe the candidate that + // confirmed, which need not be the last one broadcast. An update that doesn't assert the + // confirmation state was built without knowing it — e.g. a late funding classification + // whose candidate lost to the counterparty's broadcast — so it must not move them. + let keep_confirmed_figures = update.confirmation_status.is_none() + && matches!( + self.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + ); + + if !keep_confirmed_figures { + if let Some(amount_opt) = update.amount_msat { + update_if_necessary!(self.amount_msat, amount_opt); + } - if let Some(fee_paid_msat_opt) = update.fee_paid_msat { - update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt); + if let Some(fee_paid_msat_opt) = update.fee_paid_msat { + update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt); + } } if let Some(skimmed_fee_msat) = update.counterparty_skimmed_fee_msat { @@ -278,7 +290,7 @@ impl StorableObject for PaymentDetails { if let Some(tx_id) = update.txid { match self.kind { - PaymentKind::Onchain { ref mut txid, .. } => { + PaymentKind::Onchain { ref mut txid, .. } if !keep_confirmed_figures => { update_if_necessary!(*txid, tx_id); }, _ => {}, @@ -719,16 +731,16 @@ impl PaymentDetailsUpdate { /// [`ConfirmationStatus`] untouched. /// /// Funding classification runs off the broadcaster queue and can land *after* wallet sync has - /// already advanced a record's confirmation state (e.g. when LDK re-broadcasts a still-pending - /// funding transaction on restart, or when the counterparty's broadcast is observed first). - /// Merging only the funding-specific fields keeps such a late classification from downgrading a - /// `Confirmed`/`Succeeded` payment back to `Unconfirmed`/`Pending`; the confirmation state is - /// owned by the wallet-sync events instead. + /// already advanced a record's confirmation state (e.g. when the counterparty's broadcast of + /// the funding transaction is observed first). Merging only the funding-specific fields keeps + /// such a late classification from downgrading a `Confirmed`/`Succeeded` payment back to + /// `Unconfirmed`/`Pending`; the confirmation state is owned by the wallet-sync events instead. /// - /// The txid and figures are taken from the freshly broadcast (active) candidate. LDK only - /// re-broadcasts the active/confirmed funding candidate, so for an already-confirmed record - /// these equal what graduation stamped and the overwrite is a no-op; we rely on that invariant - /// rather than gating the txid/amount/fee merge on the stored confirmation state. + /// The txid and figures are taken from the freshly broadcast (active) candidate, so they only + /// apply while the record is unconfirmed. Once a candidate confirms, the record's txid and + /// figures describe that candidate — which need not be the one being classified (e.g. the + /// counterparty broadcast an earlier candidate and it won) — and [`PaymentDetails::update`] + /// leaves them in place for updates like this one that don't carry a confirmation state. pub(crate) fn funding_reclassification(details: PaymentDetails) -> Self { let mut update = Self::new(details.id); update.amount_msat = Some(details.amount_msat); @@ -1130,11 +1142,95 @@ mod tests { ), "reclassification must preserve the confirmation status and keep the funding tx_type", ); - // The contribution-derived figures from the fresh classification ARE merged in, replacing - // the existing record's: they are authoritative (the wallet can't recompute our share of a - // shared funding output), so the merge must carry them. - assert_eq!(merged.amount_msat, Some(1_000_000)); - assert_eq!(merged.fee_paid_msat, Some(500)); + // The confirmed record's figures describe the candidate that confirmed, so the late + // classification must not replace them either. + assert_eq!(merged.amount_msat, Some(2_000_000)); + assert_eq!(merged.fee_paid_msat, Some(999)); + } + + #[test] + fn funding_reclassification_keeps_confirmed_candidate_figures() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // A funding payment whose first candidate wallet sync has already seen confirm — e.g. the + // counterparty's broadcast of it was picked up before our own later candidate was + // classified. The record is unclassified (created by the sync fallthrough). + let confirmed_txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(confirmed_txid.to_byte_array()); + let confirmed = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // Our own, different (e.g. fee-bumped) candidate is classified late. + let late_txid = Txid::from_byte_array([9u8; 32]); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let late = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: late_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type, + }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The confirmed record's txid and figures describe the candidate that confirmed; the late + // classification must not replace them with an unconfirmed candidate's. The + // classification itself (`tx_type`) still lands. + let mut classified = confirmed.clone(); + classified.update(PaymentDetailsUpdate::funding_reclassification(late.clone())); + assert!( + matches!( + classified.kind, + PaymentKind::Onchain { + txid, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } if txid == confirmed_txid + ), + "a late classification must set the tx_type but not replace a confirmed record's txid", + ); + assert_eq!(classified.amount_msat, Some(2_000_000)); + assert_eq!(classified.fee_paid_msat, Some(999)); + + // While the record is still unconfirmed, the freshly broadcast candidate is the active + // one, so its txid and figures do replace the stored ones (RBF rotation). + let mut unconfirmed = confirmed.clone(); + if let PaymentKind::Onchain { ref mut status, .. } = unconfirmed.kind { + *status = ConfirmationStatus::Unconfirmed; + } + unconfirmed.update(PaymentDetailsUpdate::funding_reclassification(late)); + assert!( + matches!(unconfirmed.kind, PaymentKind::Onchain { txid, .. } if txid == late_txid), + "classifying a new candidate of an unconfirmed record rotates the txid", + ); + assert_eq!(unconfirmed.amount_msat, Some(1_000_000)); + assert_eq!(unconfirmed.fee_paid_msat, Some(500)); } #[tokio::test] @@ -1159,7 +1255,8 @@ mod tests { channel_id: ChannelId([3u8; 32]), }], }); - // A funding payment wallet sync has already advanced to Succeeded/Confirmed. + // A funding payment wallet sync has already recorded (unclassified, via the default + // on-chain path) and advanced to Succeeded/Confirmed. let advanced = PaymentDetails::new( id, PaymentKind::Onchain { @@ -1169,7 +1266,7 @@ mod tests { height: 100, timestamp: 1, }, - tx_type: tx_type.clone(), + tx_type: None, }, Some(2_000_000), Some(999), @@ -1210,8 +1307,8 @@ mod tests { ); // `update_or_insert` applies only the narrow reclassification when a record exists — no - // matter when it appeared — preserving the confirmation state wallet sync owns while - // still merging the contribution-derived figures. + // matter when it appeared — setting the `tx_type` while preserving the confirmation + // state wallet sync owns and the confirmed candidate's figures. let store = new_store(vec![advanced.clone()]); let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); assert_eq!( @@ -1222,10 +1319,14 @@ mod tests { assert_eq!(merged.status, PaymentStatus::Succeeded); assert!(matches!( merged.kind, - PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + PaymentKind::Onchain { + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } )); - assert_eq!(merged.amount_msat, Some(1_000_000)); - assert_eq!(merged.fee_paid_msat, Some(500)); + assert_eq!(merged.amount_msat, Some(2_000_000)); + assert_eq!(merged.fee_paid_msat, Some(999)); // And it inserts the fresh details when no record exists yet. let store = new_store(Vec::new()); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 2e6c41a436..9e5039bfdb 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1405,9 +1405,9 @@ impl Wallet { // the write, and a full merge of the fresh Pending/Unconfirmed details would then // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds // the store's mutation lock across the whole decision: when a record exists — no matter - // when it appeared — only the classification (`tx_type`), the broadcast txid, and our - // contribution figures are merged, which the wallet can't recompute; otherwise the fresh - // details are inserted. + // when it appeared — only the classification (`tx_type`) and, while the record is still + // unconfirmed, the broadcast txid and our contribution figures are merged; otherwise the + // fresh details are inserted. let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); let pending_update = PendingPaymentDetailsUpdate { id: update.id, From 435caa6b8e58e289f905eeee9c6aa1f73fb87125 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 10:50:44 -0500 Subject: [PATCH 04/11] f - Merge figures when the classified candidate is the confirmed one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The keep-confirmed-figures guard overshot: when wallet sync recorded the confirmation first, the record carries the wallet's own view of amount/fee, which cannot represent our contribution to a shared funding output, and the guard then discarded the late classification's correct contribution-derived figures along with the losing-candidate updates it was meant to block. Let an update that names the confirmed txid move the figures — it describes the very candidate that confirmed — and have classification build its update from the candidate matching an already-confirmed record, mirroring what apply_funding_status_update reports when confirmation arrives after classification. The txid comparison happens under the store's mutation lock at apply time, so the unlocked snapshot read cannot misapply figures. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/payment/pending_payment_store.rs | 7 +- src/payment/store.rs | 92 +++++++++++++++-- src/wallet/mod.rs | 145 ++++++++++++++++++++++++++- 3 files changed, 228 insertions(+), 16 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 2257de43cb..a5994c8808 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -296,7 +296,8 @@ mod tests { ); // The narrow classification update merges the candidates while preserving the - // confirmation state wallet sync owns and the confirmed candidate's figures. + // confirmation state wallet sync owns. It names the confirmed txid, so its + // contribution-derived figures replace the mirrored wallet-view ones. let mut merged = mirrored.clone(); let narrow_update = PendingPaymentDetailsUpdate { id: payment_id, @@ -313,7 +314,7 @@ mod tests { "a narrow classification update must not downgrade a mirrored confirmation", ); assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(2_000_000)); - assert_eq!(merged.details.fee_paid_msat, Some(999)); + assert_eq!(merged.details.amount_msat, Some(1_000)); + assert_eq!(merged.details.fee_paid_msat, Some(100)); } } diff --git a/src/payment/store.rs b/src/payment/store.rs index 00fcffcff8..8eb787286c 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -246,11 +246,15 @@ impl StorableObject for PaymentDetails { // Once an on-chain record is confirmed, its txid and figures describe the candidate that // confirmed, which need not be the last one broadcast. An update that doesn't assert the // confirmation state was built without knowing it — e.g. a late funding classification - // whose candidate lost to the counterparty's broadcast — so it must not move them. + // whose candidate lost to the counterparty's broadcast — so it must not move them. The + // exception is an update naming the confirmed txid itself: its figures describe the very + // candidate that confirmed and correct the wallet-view amount/fee a sync-created record + // carries, which cannot represent our contribution to a shared funding output. let keep_confirmed_figures = update.confirmation_status.is_none() && matches!( self.kind, - PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + PaymentKind::Onchain { txid, status: ConfirmationStatus::Confirmed { .. }, .. } + if update.txid != Some(txid) ); if !keep_confirmed_figures { @@ -1142,10 +1146,11 @@ mod tests { ), "reclassification must preserve the confirmation status and keep the funding tx_type", ); - // The confirmed record's figures describe the candidate that confirmed, so the late - // classification must not replace them either. - assert_eq!(merged.amount_msat, Some(2_000_000)); - assert_eq!(merged.fee_paid_msat, Some(999)); + // The late classification names the confirmed txid, so its contribution-derived figures + // replace the record's; only an update for a different candidate leaves them in place + // (covered by `funding_reclassification_keeps_confirmed_candidate_figures`). + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); } #[test] @@ -1233,6 +1238,74 @@ mod tests { assert_eq!(unconfirmed.fee_paid_msat, Some(500)); } + #[test] + fn funding_reclassification_merges_figures_for_the_confirmed_candidate() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // Wallet sync confirmed the transaction before classification ran, so the record carries + // the wallet's own view of amount/fee, which cannot represent our contribution to a shared + // funding output. + let confirmed_txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(confirmed_txid.to_byte_array()); + let mut record = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The late classification names the candidate that confirmed, so its contribution-derived + // figures are authoritative and must replace the wallet-view ones; only an update for a + // different (losing) candidate leaves a confirmed record's figures in place. + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let classified = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type, + }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + assert!(record.update(PaymentDetailsUpdate::funding_reclassification(classified))); + assert!( + matches!( + record.kind, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + } if txid == confirmed_txid + ), + "the confirmed txid, confirmation state, and classification must all be in place", + ); + assert_eq!(record.amount_msat, Some(1_000_000)); + assert_eq!(record.fee_paid_msat, Some(500)); + } + #[tokio::test] async fn funding_classification_update_or_insert_preserves_advanced_record() { use bitcoin::hashes::Hash; @@ -1308,7 +1381,8 @@ mod tests { // `update_or_insert` applies only the narrow reclassification when a record exists — no // matter when it appeared — setting the `tx_type` while preserving the confirmation - // state wallet sync owns and the confirmed candidate's figures. + // state wallet sync owns. The update names the confirmed txid, so its + // contribution-derived figures replace the record's wallet-view ones. let store = new_store(vec![advanced.clone()]); let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); assert_eq!( @@ -1325,8 +1399,8 @@ mod tests { .. } )); - assert_eq!(merged.amount_msat, Some(2_000_000)); - assert_eq!(merged.fee_paid_msat, Some(999)); + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); // And it inserts the fresh details when no record exists yet. let store = new_store(Vec::new()); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 9e5039bfdb..a4c1268fd1 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1405,10 +1405,14 @@ impl Wallet { // the write, and a full merge of the fresh Pending/Unconfirmed details would then // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds // the store's mutation lock across the whole decision: when a record exists — no matter - // when it appeared — only the classification (`tx_type`) and, while the record is still - // unconfirmed, the broadcast txid and our contribution figures are merged; otherwise the - // fresh details are inserted. - let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); + // when it appeared — only the classification (`tx_type`) and the figures of whichever + // candidate the record's state makes authoritative are merged; otherwise the fresh + // details are inserted. + let update = funding_reclassification_update( + details.clone(), + &candidates, + self.payment_store.get(&details.id).as_ref(), + ); let pending_update = PendingPaymentDetailsUpdate { id: update.id, payment_update: Some(update.clone()), @@ -2187,3 +2191,136 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { .saturating_sub(EMPTY_SCRIPT_SIG_WEIGHT + EMPTY_WITNESS_COUNT_WEIGHT), ) } + +/// Builds the payment-store update for a freshly classified funding payment. `details` describes +/// the actively broadcast candidate, but when the record already confirmed a *different* +/// candidate — wallet sync saw it win before this classification ran — the update instead carries +/// the confirmed candidate's txid and figures from the candidate history, mirroring what +/// [`Wallet::apply_funding_status_update`] reports when confirmation arrives after classification. +/// +/// `current` is an unlocked snapshot; that is safe because [`PaymentDetails::update`] only lets +/// figures onto a confirmed record when the update names the confirmed txid, so an update built +/// against a stale snapshot cannot misapply figures if the record's confirmation moves before the +/// update lands. +fn funding_reclassification_update( + details: PaymentDetails, candidates: &[FundingTxCandidate], current: Option<&PaymentDetails>, +) -> PaymentDetailsUpdate { + let mut update = PaymentDetailsUpdate::funding_reclassification(details); + if let Some(PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { .. }, + .. + }) = current.map(|payment| &payment.kind) + { + if update.txid != Some(*confirmed_txid) { + if let Some(candidate) = candidates.iter().find(|c| c.txid == *confirmed_txid) { + update.txid = Some(candidate.txid); + update.amount_msat = Some(candidate.amount_msat); + update.fee_paid_msat = Some(candidate.fee_paid_msat); + } + } + } + update +} + +#[cfg(test)] +mod tests { + use bitcoin::hashes::Hash; + + use super::*; + + fn onchain_details(txid: Txid, status: ConfirmationStatus) -> PaymentDetails { + PaymentDetails::new( + PaymentId([42u8; 32]), + PaymentKind::Onchain { txid, status, tx_type: None }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ) + } + + fn confirmed_status() -> ConfirmationStatus { + ConfirmationStatus::Confirmed { + block_hash: bitcoin::BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + } + } + + #[test] + fn funding_reclassification_update_substitutes_the_confirmed_candidate() { + let confirmed_txid = Txid::from_byte_array([1u8; 32]); + let active_txid = Txid::from_byte_array([2u8; 32]); + let candidates = vec![ + FundingTxCandidate { + txid: confirmed_txid, + amount_msat: Some(2_000_000), + fee_paid_msat: Some(999), + }, + FundingTxCandidate { + txid: active_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + ]; + let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed); + + // The record confirmed an earlier candidate: the update reports that candidate, not the + // active one. + let current = onchain_details(confirmed_txid, confirmed_status()); + let update = funding_reclassification_update(details.clone(), &candidates, Some(¤t)); + assert_eq!(update.txid, Some(confirmed_txid)); + assert_eq!(update.amount_msat, Some(Some(2_000_000))); + assert_eq!(update.fee_paid_msat, Some(Some(999))); + + // A confirmed candidate we did not contribute to still substitutes, with empty figures — + // the same figures a confirmation arriving after classification would report. + let uncontributed = vec![FundingTxCandidate { + txid: confirmed_txid, + amount_msat: None, + fee_paid_msat: None, + }]; + let update = + funding_reclassification_update(details.clone(), &uncontributed, Some(¤t)); + assert_eq!(update.txid, Some(confirmed_txid)); + assert_eq!(update.amount_msat, Some(None)); + assert_eq!(update.fee_paid_msat, Some(None)); + } + + #[test] + fn funding_reclassification_update_keeps_the_active_candidate() { + 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 details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed); + + // No record yet: the update describes the active candidate. + let update = funding_reclassification_update(details.clone(), &candidates, None); + 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); + 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)); + 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)); + } +} From 6168bb019cdfcec8ae4b95e45dd9cde0eacc2234 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 10:52:46 -0500 Subject: [PATCH 05/11] f - Recreate a missing pending index while the payment is still Pending Classification treated an absent pending entry as proof the payment had graduated and only merged into existing entries. But a crash or failed write between the payment-store and pending-store writes leaves a Pending record with no index entry, and that state was never repaired: the payment could no longer graduate (graduation iterates the pending store) and its candidate txids could no longer be mapped back to the record, which for an RBF splice invites a duplicate generic payment. Decide by the post-write payment status instead: while the record is still Pending, insert the missing entry (embedding the post-write record, so a confirmation wallet sync already mirrored keeps driving graduation); once it advanced beyond Pending, keep treating absence as graduated. A graduated payment is never Pending, so the no-reindex rule is preserved by the status gate itself. The repaired state is not constructible in a test: it requires a failure injected between the two store writes, and no such seam exists. The store primitives the decision rests on are unit-tested. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 40 +++++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index a4c1268fd1..ca4674d565 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,7 +54,6 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; -use crate::data_store::DataStoreUpdateOrInsertResult; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -1419,24 +1418,27 @@ impl Wallet { conflicting_txids: None, candidates: candidates.clone(), }; - match self.payment_store.update_or_insert(update, details.clone()).await? { - DataStoreUpdateOrInsertResult::Inserted => { - // First time we record this funding payment: index it for graduation. Wallet sync - // can still land between the payment-store write above and this one and mirror an - // advanced confirmation into the pending store, so this write makes the same - // atomic decision: merge narrowly into an entry that appeared, insert the fresh - // one otherwise. - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.update_or_insert(pending_update, pending).await?; - }, - DataStoreUpdateOrInsertResult::Updated | DataStoreUpdateOrInsertResult::Unchanged => { - // An earlier candidate or a racing wallet sync already recorded this payment. - // `update` is a no-op when the pending entry is absent, so the index is not - // re-created for a payment the graduation path already removed. (A graduated - // payment always has a payment-store record, so it cannot take the `Inserted` - // branch above and be re-indexed.) - self.pending_payment_store.update(pending_update).await?; - }, + self.payment_store.update_or_insert(update, details.clone()).await?; + + // The pending index must exist exactly while the authoritative record is Pending: + // graduation and rebroadcast read it, and a graduated payment must not be re-indexed. + // Deciding by the post-write status rather than by whether the write inserted also + // repairs a missing index — a crash or failed write between the two stores leaves a + // Pending record with no entry, and a merge alone would never recreate it, leaving the + // payment unable to graduate and its txids unmapped. + let recorded = self.payment_store.get(&details.id).unwrap_or(details); + if recorded.status == PaymentStatus::Pending { + // Wallet sync can still land between the payment-store write above and this one and + // mirror an advanced confirmation into the pending store, so this write makes the + // same atomic decision: merge narrowly into an entry that appeared, insert otherwise. + // The inserted entry embeds the post-write record rather than the fresh details, so a + // confirmation wallet sync already recorded keeps driving graduation. + let pending = PendingPaymentDetails::new(recorded, Vec::new(), candidates); + self.pending_payment_store.update_or_insert(pending_update, pending).await?; + } else { + // The payment already advanced beyond Pending: the graduation path removed the + // entry, and `update`'s no-op on absence must not re-create it. + self.pending_payment_store.update(pending_update).await?; } Ok(()) } From 1192866586a0d89bc683304ee7bcf104ea410d57 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 1 Jul 2026 23:35:15 -0500 Subject: [PATCH 06/11] Model pending payments as an enum for pre-broadcast splices Retrying a user-initiated splice across restarts requires persisting the splice intent before handing it to LDK, which happens before negotiation and therefore before any funding transaction exists. The pending-payment record was built around an on-chain PaymentDetails carrying a txid, which cannot represent a splice that has not been broadcast yet. Reshape PendingPaymentDetails into an enum: a PendingSplice variant that holds only the generated PaymentId and the splice intent, and a Tracked variant that is the previous record plus an optional intent retained until the splice locks. Add the SpliceIntent and SpliceKind types the intent needs to resubmit or rebuild the contribution. Wallet writes to the pending store go through a new DataStore::mutate primitive that reads, transforms, and persists an entry under one critical section of the mutation lock, replacing racy read-then-write pairs. The wallet's pending-store writes share one helper whose closure promotes a bare PendingSplice to a Tracked record once a payment exists under its id: a plain payment-tracking merge would silently no-op against the variant, leaving the splice invisible to txid lookups. This is groundwork; nothing constructs a PendingSplice yet. The classify, retry, and wiring that use it follow in subsequent commits. Generated with assistance from Claude Code. Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Fable 5 --- src/data_store.rs | 126 ++++++++++ src/payment/pending_payment_store.rs | 339 ++++++++++++++++++++++----- src/wallet/mod.rs | 108 +++++---- 3 files changed, 468 insertions(+), 105 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index b7748b172e..f259936d03 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -223,6 +223,35 @@ where Ok(result) } + /// Atomically transforms the entry for `id` through `f` and persists the result. + /// + /// `f` receives the current entry (`None` when absent) and returns the new state to write; + /// returning `None` leaves the store untouched. The read, the closure, and the write share one + /// critical section of the mutation lock, so no concurrent writer can land in between — unlike + /// a separate [`Self::get`] followed by an insert or update. Keep the closure cheap and + /// non-blocking, and do not call back into this store from it. + /// + /// Returns the written object, or `None` when the closure declined to write. + pub(crate) async fn mutate) -> Option>( + &self, id: &SO::Id, f: F, + ) -> Result, Error> { + let _guard = self.mutation_lock.lock().await; + + let new_object = { + let locked_objects = self.objects.lock().expect("lock"); + match f(locked_objects.get(id)) { + Some(new_object) => new_object, + None => return Ok(None), + } + }; + debug_assert!(new_object.id() == *id, "mutate closure must not change the object's id"); + + self.persist(&new_object).await?; + let mut locked_objects = self.objects.lock().expect("lock"); + locked_objects.insert(new_object.id(), new_object.clone()); + Ok(Some(new_object)) + } + /// Returns in-memory objects matching `f`. /// /// The async mutation lock serializes writers, but this synchronous reader cannot wait on it. @@ -552,6 +581,103 @@ mod tests { assert!(data_store.get(&new_id).is_none()); } + #[tokio::test] + async fn mutate_inserts_when_absent() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let primary_namespace = "datastore_test_primary".to_string(); + let secondary_namespace = "datastore_test_secondary".to_string(); + let data_store: DataStore> = DataStore::new( + Vec::new(), + primary_namespace.clone(), + secondary_namespace.clone(), + Arc::clone(&store), + logger, + ); + + let id = TestObjectId { id: [42u8; 4] }; + let object = TestObject { id, data: [23u8; 3] }; + let result = data_store + .mutate(&id, |existing| { + assert!(existing.is_none()); + Some(object) + }) + .await; + assert_eq!(Ok(Some(object)), result); + + assert_eq!(Some(object), data_store.get(&id)); + let store_key = id.encode_to_hex_str(); + assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key) + .await + .is_ok()); + } + + #[tokio::test] + async fn mutate_transforms_existing_entry() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store: DataStore> = DataStore::new( + vec![existing_object], + "datastore_test_primary".to_string(), + "datastore_test_secondary".to_string(), + store, + logger, + ); + + // The closure sees the current entry and derives the new state from it. + let result = data_store + .mutate(&id, |existing| { + let mut new_object = *existing.unwrap(); + new_object.data[0] += 1; + Some(new_object) + }) + .await; + let expected = TestObject { id, data: [24u8, 23u8, 23u8] }; + assert_eq!(Ok(Some(expected)), result); + assert_eq!(Some(expected), data_store.get(&id)); + } + + #[tokio::test] + async fn mutate_persists_nothing_when_closure_declines() { + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + // Returning `None` must not attempt a write (the store fails all writes) nor touch memory. + let result = data_store + .mutate(&id, |existing| { + assert_eq!(Some(&existing_object), existing); + None + }) + .await; + assert_eq!(Ok(None), result); + assert_eq!(Some(existing_object), data_store.get(&id)); + } + + #[tokio::test] + async fn mutate_does_not_mutate_memory_if_persist_fails() { + let existing_id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id: existing_id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + let changed = TestObject { id: existing_id, data: [24u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.mutate(&existing_id, |_| Some(changed)).await + ); + assert_eq!(Some(existing_object), data_store.get(&existing_id)); + + let new_id = TestObjectId { id: [55u8; 4] }; + let new_object = TestObject { id: new_id, data: [34u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.mutate(&new_id, |_| Some(new_object)).await + ); + assert!(data_store.get(&new_id).is_none()); + } + #[tokio::test] async fn insert_or_update_does_not_mutate_memory_if_persist_fails() { let existing_id = TestObjectId { id: [42u8; 4] }; diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index a5994c8808..05772d259b 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -5,13 +5,18 @@ // 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 bitcoin::Txid; -use lightning::impl_writeable_tlv_based; +use bitcoin::secp256k1::PublicKey; +use bitcoin::{TxOut, Txid}; +use lightning::chain::transaction::OutPoint; use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; +use lightning::ln::types::ChannelId; +use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use crate::data_store::{StorableObject, StorableObjectUpdate}; use crate::payment::store::PaymentDetailsUpdate; use crate::payment::{PaymentDetails, PaymentKind}; +use crate::types::UserChannelId; /// One candidate transaction in an interactive-funding (splice) RBF history, holding this node's /// share of the funding amount and fee for that candidate. Both are `None` for a candidate this @@ -36,37 +41,157 @@ impl_writeable_tlv_based!(FundingTxCandidate, { (4, fee_paid_msat, option), }); -/// Represents a pending payment +/// The parameters of the API call that initiated a splice, used to rebuild a fresh contribution +/// when the stored one has become stale (e.g. its feerate is no longer sufficient). #[derive(Clone, Debug, PartialEq, Eq)] -pub struct PendingPaymentDetails { - /// The full payment details - pub details: PaymentDetails, - /// Transaction IDs that have replaced or conflict with this payment. - pub conflicting_txids: Vec, - /// For interactive funding (splices), this node's per-candidate funding figures across the - /// RBF history, keyed by each candidate's txid. Empty for non-funding payments and for - /// records written before per-candidate tracking existed. - pub(crate) candidates: Vec, +pub(crate) enum SpliceKind { + /// [`Node::splice_in`] with a resolved amount. + /// + /// [`Node::splice_in`]: crate::Node::splice_in + In { amount_sats: u64 }, + /// [`Node::splice_out`] to the given outputs. + /// + /// [`Node::splice_out`]: crate::Node::splice_out + Out { outputs: Vec }, + /// [`Node::bump_channel_funding_fee`] of a pending splice. + /// + /// [`Node::bump_channel_funding_fee`]: crate::Node::bump_channel_funding_fee + Rbf {}, +} + +impl_writeable_tlv_based_enum!(SpliceKind, + (0, In) => { + (0, amount_sats, required), + }, + (2, Out) => { + (0, outputs, required_vec), + }, + (4, Rbf) => {}, +); + +/// A user-initiated splice that has been handed to LDK but is not yet guaranteed to survive a +/// restart. LDK only persists a splice once its negotiation reaches `AwaitingSignatures`, and it +/// abandons an in-progress negotiation whenever the peer disconnects (which includes stopping the +/// node). Until the new funding transaction locks we keep enough state to resubmit the splice +/// ourselves. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct SpliceIntent { + /// The channel's local identifier, carried so the retrier can address the splice without a + /// separate store keyed by it. + pub user_channel_id: UserChannelId, + /// The channel counterparty. + pub counterparty_node_id: PublicKey, + /// The channel being spliced. + pub channel_id: ChannelId, + /// The channel's funding outpoint when the splice was initiated. It only changes once a splice + /// locks, so a mismatch with the channel's current funding outpoint means the splice (or a + /// replacement) completed and there is nothing left to resubmit. + pub pre_splice_funding_txo: OutPoint, + /// The contribution handed to [`ChannelManager::funding_contributed`], resubmitted verbatim. + /// + /// [`ChannelManager::funding_contributed`]: lightning::ln::channelmanager::ChannelManager::funding_contributed + pub contribution: FundingContribution, + /// The parameters of the originating API call, used to rebuild a fresh contribution when the + /// stored one has become stale. + pub kind: SpliceKind, + /// The number of times the contribution has been resubmitted to LDK after the originating API + /// call handed it off. + pub attempts: u8, +} + +impl_writeable_tlv_based!(SpliceIntent, { + (0, user_channel_id, required), + (2, counterparty_node_id, required), + (4, channel_id, required), + (6, pre_splice_funding_txo, required), + (8, contribution, required), + (10, kind, required), + (12, attempts, required), +}); + +/// A pending payment tracked by LDK Node, keyed by [`PaymentId`]. +/// +/// A user-initiated splice is persisted as a [`PendingSplice`] before its contribution is handed +/// to LDK — at which point no funding transaction, and therefore no [`PaymentDetails`], exists yet. +/// Once the splice is broadcast and classified it becomes a [`Tracked`] payment carrying the real +/// [`PaymentDetails`], while retaining its [`SpliceIntent`] until the splice locks. +/// +/// [`PendingSplice`]: Self::PendingSplice +/// [`Tracked`]: Self::Tracked +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum PendingPaymentDetails { + /// A user-initiated splice persisted before hand-off to LDK; no funding transaction exists yet. + /// Keyed by the generated [`PaymentId`]; never mirrored into the payment store. + PendingSplice { id: PaymentId, intent: SpliceIntent }, + /// A pending payment tracked toward confirmation, optionally still carrying a live splice + /// intent to resubmit until the splice locks. + Tracked { + /// The full payment details. + details: PaymentDetails, + /// Transaction IDs that have replaced or conflict with this payment. + conflicting_txids: Vec, + /// For interactive funding (splices), this node's per-candidate funding figures across the + /// RBF history, keyed by each candidate's txid. Empty for non-funding payments. + candidates: Vec, + /// The splice intent to resubmit if LDK drops the splice before it locks, or `None` for a + /// non-splice payment or a splice that has locked. + splice_intent: Option, + }, } impl PendingPaymentDetails { pub(crate) fn new( details: PaymentDetails, conflicting_txids: Vec, candidates: Vec, ) -> Self { - Self { details, conflicting_txids, candidates } + Self::tracked(details, conflicting_txids, candidates, None) + } + + pub(crate) fn tracked( + details: PaymentDetails, conflicting_txids: Vec, candidates: Vec, + splice_intent: Option, + ) -> Self { + Self::Tracked { details, conflicting_txids, candidates, splice_intent } + } + + /// The full payment details, or `None` for a splice not yet broadcast. + pub(crate) fn details(&self) -> Option<&PaymentDetails> { + match self { + Self::PendingSplice { .. } => None, + Self::Tracked { details, .. } => Some(details), + } + } + + /// Transaction IDs that have replaced or conflict with this payment. + pub(crate) fn conflicting_txids(&self) -> &[Txid] { + match self { + Self::PendingSplice { .. } => &[], + Self::Tracked { conflicting_txids, .. } => conflicting_txids, + } } /// Returns this node's recorded funding figures for the candidate with the given txid, if any. pub(crate) fn candidate(&self, txid: Txid) -> Option<&FundingTxCandidate> { - self.candidates.iter().find(|candidate| candidate.txid == txid) + match self { + Self::PendingSplice { .. } => None, + Self::Tracked { candidates, .. } => { + candidates.iter().find(|candidate| candidate.txid == txid) + }, + } } } -impl_writeable_tlv_based!(PendingPaymentDetails, { - (0, details, required), - (2, conflicting_txids, optional_vec), - (4, candidates, optional_vec), -}); +impl_writeable_tlv_based_enum!(PendingPaymentDetails, + (0, PendingSplice) => { + (0, id, required), + (2, intent, required), + }, + (2, Tracked) => { + (0, details, required), + (2, conflicting_txids, optional_vec), + (4, candidates, optional_vec), + (6, splice_intent, option), + }, +); #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct PendingPaymentDetailsUpdate { @@ -74,6 +199,11 @@ pub(crate) struct PendingPaymentDetailsUpdate { pub payment_update: Option, pub conflicting_txids: Option>, pub candidates: Vec, + /// The splice intent to set (`Some(Some(..))`) or clear (`Some(None)`), or `None` to leave it + /// unchanged. Setting it on a [`PendingPaymentDetails::PendingSplice`] replaces the intent (e.g. + /// to bump the retry attempt count); clearing a pre-broadcast splice is done by removing the + /// record, not through this field. + pub splice_intent: Option>, } impl StorableObject for PendingPaymentDetails { @@ -81,38 +211,64 @@ impl StorableObject for PendingPaymentDetails { type Update = PendingPaymentDetailsUpdate; fn id(&self) -> Self::Id { - self.details.id + match self { + Self::PendingSplice { id, .. } => *id, + Self::Tracked { details, .. } => details.id, + } } fn update(&mut self, update: Self::Update) -> bool { - let mut updated = false; - - // Update the underlying payment details if present - if let Some(payment_update) = update.payment_update { - updated |= self.details.update(payment_update); - } - - if let Some(new_conflicting_txids) = update.conflicting_txids { - if self.conflicting_txids != new_conflicting_txids { - self.conflicting_txids = new_conflicting_txids; - updated = true; - } - } - - if let PaymentKind::Onchain { txid, .. } = &self.details.kind { - let conflicts_len = self.conflicting_txids.len(); - self.conflicting_txids.retain(|conflicting_txid| conflicting_txid != txid); - 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 { - self.candidates = update.candidates; - updated = true; + match self { + Self::PendingSplice { intent, .. } => { + // A pre-broadcast record only carries a splice intent; the only meaningful update is + // replacing that intent. Clearing it is done by removing the record. + if let Some(Some(new_intent)) = update.splice_intent { + if *intent != new_intent { + *intent = new_intent; + return true; + } + } + false + }, + Self::Tracked { details, conflicting_txids, candidates, splice_intent } => { + let mut updated = false; + + // Update the underlying payment details if present + if let Some(payment_update) = update.payment_update { + updated |= details.update(payment_update); + } + + if let Some(new_conflicting_txids) = update.conflicting_txids { + if *conflicting_txids != new_conflicting_txids { + *conflicting_txids = new_conflicting_txids; + updated = true; + } + } + + if let PaymentKind::Onchain { txid, .. } = &details.kind { + let conflicts_len = conflicting_txids.len(); + conflicting_txids.retain(|conflicting_txid| conflicting_txid != txid); + updated |= 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() && *candidates != update.candidates { + *candidates = update.candidates; + updated = true; + } + + if let Some(new_splice_intent) = update.splice_intent { + if *splice_intent != new_splice_intent { + *splice_intent = new_splice_intent; + updated = true; + } + } + + updated + }, } - - updated } fn to_update(&self) -> Self::Update { @@ -128,16 +284,33 @@ impl StorableObjectUpdate for PendingPaymentDetailsUpdate impl From<&PendingPaymentDetails> for PendingPaymentDetailsUpdate { fn from(value: &PendingPaymentDetails) -> Self { - let conflicting_txids = if value.conflicting_txids.is_empty() { - None - } else { - Some(value.conflicting_txids.clone()) - }; - Self { - id: value.id(), - payment_update: Some(value.details.to_update()), - conflicting_txids, - candidates: value.candidates.clone(), + match value { + PendingPaymentDetails::PendingSplice { id, intent } => Self { + id: *id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent.clone())), + }, + PendingPaymentDetails::Tracked { + details, + conflicting_txids, + candidates, + splice_intent, + } => { + let conflicting_txids = if conflicting_txids.is_empty() { + None + } else { + Some(conflicting_txids.clone()) + }; + Self { + id: details.id, + payment_update: Some(details.to_update()), + conflicting_txids, + candidates: candidates.clone(), + splice_intent: Some(splice_intent.clone()), + } + }, } } } @@ -145,6 +318,7 @@ impl From<&PendingPaymentDetails> for PendingPaymentDetailsUpdate { #[cfg(test)] mod tests { use bitcoin::hashes::Hash; + use lightning::util::ser::{Readable, Writeable}; use super::*; use crate::payment::store::ConfirmationStatus; @@ -237,7 +411,7 @@ mod tests { assert!(pending_payment.update(update)); assert_eq!( - pending_payment.conflicting_txids, + pending_payment.conflicting_txids(), Vec::::new(), "current txid must not remain in its own conflict list" ); @@ -289,7 +463,7 @@ mod tests { assert!(downgraded.update(full_update)); assert!( matches!( - downgraded.details.kind, + downgraded.details().expect("tracked").kind, PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } ), "a full merge of a fresh classification downgrades a mirrored confirmation", @@ -304,17 +478,56 @@ mod tests { payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)), conflicting_txids: None, candidates: candidates.clone(), + splice_intent: None, }; assert!(merged.update(narrow_update)); + let merged_details = merged.details().expect("tracked"); assert!( matches!( - merged.details.kind, + merged_details.kind, PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } ), "a narrow classification update must not downgrade a mirrored confirmation", ); - assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(1_000)); - assert_eq!(merged.details.fee_paid_msat, Some(100)); + assert_eq!(merged.candidate(txid), Some(&candidates[0])); + assert_eq!(merged_details.amount_msat, Some(1_000)); + assert_eq!(merged_details.fee_paid_msat, Some(100)); + } + + #[test] + fn splice_kind_round_trips() { + for kind in [ + SpliceKind::In { amount_sats: 500_000 }, + SpliceKind::Out { + outputs: vec![TxOut { + value: bitcoin::Amount::from_sat(400_000), + script_pubkey: bitcoin::ScriptBuf::new(), + }], + }, + SpliceKind::Rbf {}, + ] { + let encoded = kind.encode(); + let decoded = SpliceKind::read(&mut &encoded[..]).unwrap(); + assert_eq!(kind, decoded); + } + } + + #[test] + fn tracked_payment_round_trips() { + // A `PendingSplice` record round-trips through the restart integration tests, which persist a + // real `FundingContribution`; here we cover the `Tracked` variant and its enum discriminant. + let payment_id = PaymentId([7u8; 32]); + let txid = Txid::from_byte_array([8u8; 32]); + let record = PendingPaymentDetails::new( + pending_onchain_payment(payment_id, txid), + vec![Txid::from_byte_array([9u8; 32])], + vec![FundingTxCandidate { txid, amount_msat: Some(1_000), fee_paid_msat: Some(100) }], + ); + + let encoded = record.encode(); + let decoded = PendingPaymentDetails::read(&mut &encoded[..]).unwrap(); + assert_eq!(record, decoded); + assert_eq!(decoded.id(), payment_id); + assert!(decoded.details().is_some()); } } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index ca4674d565..b309436f75 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; +use crate::data_store::StorableObject; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -282,36 +283,41 @@ impl Wallet { self.payment_store.insert_or_update(payment.clone()).await?; if payment_status == PaymentStatus::Pending { - let pending_payment = - self.create_pending_payment_from_tx(payment, Vec::new()); - - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; } }, WalletEvent::ChainTipChanged { new_tip, .. } => { let pending_payments: Vec = - self.pending_payment_store.list_filter(|p| { - debug_assert!( - p.details.status == PaymentStatus::Pending, - "Non-pending payment {:?} found in pending store", - p.details.id, - ); - p.details.status == PaymentStatus::Pending - && matches!(p.details.kind, PaymentKind::Onchain { .. }) + self.pending_payment_store.list_filter(|p| match p.details() { + // A pre-broadcast splice intent carries no payment yet and cannot graduate. + None => false, + Some(details) => { + debug_assert!( + details.status == PaymentStatus::Pending, + "Non-pending payment {:?} found in pending store", + details.id, + ); + details.status == PaymentStatus::Pending + && matches!(details.kind, PaymentKind::Onchain { .. }) + }, }); let mut unconfirmed_outbound_txids: Vec = Vec::new(); - for mut payment in pending_payments { - match payment.details.kind { + for payment in pending_payments { + // The filter admits only Tracked funding payments. + let PendingPaymentDetails::Tracked { mut details, .. } = payment else { + continue; + }; + match details.kind { PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { height, .. }, .. } => { - let payment_id = payment.details.id; + let payment_id = details.id; if new_tip.height >= height + ANTI_REORG_DELAY - 1 { - payment.details.status = PaymentStatus::Succeeded; - self.payment_store.insert_or_update(payment.details).await?; + details.status = PaymentStatus::Succeeded; + self.payment_store.insert_or_update(details).await?; self.pending_payment_store.remove(&payment_id).await?; } }, @@ -319,7 +325,7 @@ impl Wallet { txid, status: ConfirmationStatus::Unconfirmed, .. - } if payment.details.direction == PaymentDirection::Outbound => { + } if details.direction == PaymentDirection::Outbound => { unconfirmed_outbound_txids.push(txid); }, _ => {}, @@ -379,10 +385,8 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ) }; - let pending_payment = - self.create_pending_payment_from_tx(payment.clone(), Vec::new()); - self.payment_store.insert_or_update(payment).await?; - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.payment_store.insert_or_update(payment.clone()).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; }, WalletEvent::TxReplaced { txid, conflicts, .. } => { let Some(payment_id) = self.find_payment_by_txid(txid) else { @@ -409,10 +413,7 @@ impl Wallet { ); let payment = self.payment_store.get(&payment_id).ok_or(Error::InvalidPaymentId)?; - let pending_payment_details = - self.create_pending_payment_from_tx(payment, conflict_txids.clone()); - - self.pending_payment_store.insert_or_update(pending_payment_details).await?; + self.upsert_pending_payment(payment, conflict_txids).await?; }, WalletEvent::TxDropped { txid, tx } => { let payment_id = self @@ -441,10 +442,8 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ) }; - let pending_payment = - self.create_pending_payment_from_tx(payment.clone(), Vec::new()); - self.payment_store.insert_or_update(payment).await?; - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.payment_store.insert_or_update(payment.clone()).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; }, _ => { continue; @@ -1417,6 +1416,7 @@ impl Wallet { payment_update: Some(update.clone()), conflicting_txids: None, candidates: candidates.clone(), + splice_intent: None, }; self.payment_store.update_or_insert(update, details.clone()).await?; @@ -1505,10 +1505,36 @@ impl Wallet { PaymentDetails::new(payment_id, kind, amount_msat, fee_paid_msat, direction, payment_status) } - fn create_pending_payment_from_tx( + /// Inserts or refreshes the pending-store entry tracking `payment` toward graduation, + /// atomically with reading the entry's current state. Only `Pending` payments belong in the + /// pending store; callers gate on that. + async fn upsert_pending_payment( &self, payment: PaymentDetails, conflicting_txids: Vec, - ) -> PendingPaymentDetails { - PendingPaymentDetails::new(payment, conflicting_txids, Vec::new()) + ) -> Result<(), Error> { + let id = payment.id; + self.pending_payment_store + .mutate(&id, |existing| match existing { + None => Some(PendingPaymentDetails::new(payment, conflicting_txids, Vec::new())), + // Promote a pre-broadcast splice intent: wallet sync saw the splice transaction + // before its broadcast-time classification recorded it. Carrying the intent into + // the `Tracked` record makes the entry visible to txid lookups while the retrier + // keeps the intent until the splice locks. + Some(PendingPaymentDetails::PendingSplice { intent, .. }) => { + Some(PendingPaymentDetails::tracked( + payment, + conflicting_txids, + Vec::new(), + Some(intent.clone()), + )) + }, + Some(tracked @ PendingPaymentDetails::Tracked { .. }) => { + let mut updated = tracked.clone(); + let fresh = PendingPaymentDetails::new(payment, conflicting_txids, Vec::new()); + updated.update(fresh.to_update()).then_some(updated) + }, + }) + .await?; + Ok(()) } fn find_payment_by_txid(&self, target_txid: Txid) -> Option { @@ -1520,12 +1546,13 @@ impl Wallet { if let Some(replaced_details) = self .pending_payment_store .list_filter(|p| { - matches!(p.details.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid) - || p.conflicting_txids.contains(&target_txid) + p.details().is_some_and( + |d| matches!(d.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid), + ) || p.conflicting_txids().contains(&target_txid) }) .first() { - return Some(replaced_details.details.id); + return Some(replaced_details.id()); } None @@ -1573,8 +1600,7 @@ impl Wallet { // the same dual-write the default `TxConfirmed` path performs; an empty conflicting-txids // list leaves any stored conflicts intact (the update treats absent as "unchanged"). if payment.status == PaymentStatus::Pending { - let pending = self.create_pending_payment_from_tx(payment, Vec::new()); - self.pending_payment_store.insert_or_update(pending).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; } Ok(true) } @@ -1817,8 +1843,6 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ); - let pending_payment_store = - self.create_pending_payment_from_tx(new_payment.clone(), Vec::new()); let change_set = locked_wallet.take_staged().unwrap_or_default(); drop(locked_wallet); locked_persister.persist_changeset(change_set).await.map_err(|e| { @@ -1826,8 +1850,8 @@ impl Wallet { Error::PersistenceFailed })?; - self.payment_store.insert_or_update(new_payment).await?; - self.pending_payment_store.insert_or_update(pending_payment_store).await?; + self.payment_store.insert_or_update(new_payment.clone()).await?; + self.upsert_pending_payment(new_payment, Vec::new()).await?; self.broadcaster.broadcast_unclassified_transaction(fee_bumped_tx); From f6bcf4fcf4077419d9c416013c617153432e4034 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 1 Jul 2026 23:40:00 -0500 Subject: [PATCH 07/11] Adopt the splice-time PaymentId when classifying a splice A user-initiated splice will be keyed by a PaymentId generated at splice time rather than derived from a candidate's txid, so its retry intent, funding payment, and candidate history all share one record. Teach the classifier to find a pre-broadcast splice intent by its channel and reuse that id, promoting the intent record to a tracked funding payment while preserving the intent until the splice locks. Splices we did not originate (counterparty-initiated or V2 dual-funded opens) keep deriving the id from the first candidate's txid via a fallback. Also map any candidate txid back to the record in find_payment_by_txid, since a splice under a generated id is no longer found by the txid-derived lookup and an earlier RBF candidate may be the one that confirms. If the intent is already gone when classification runs, the classifier probes those same lookups for a record any candidate already created before minting a txid-derived id, so a racing wallet sync and a late classification converge on one record. The pending-store write in persist_funding_payment now happens atomically with reading the entry's prior state, and promotion of a pre-broadcast intent is gated on the payment still being Pending: a payment that confirmed through ANTI_REORG_DELAY before classification must not re-enter the pending store, which graduation and rebroadcast assume holds only Pending payments. No splice intents are created yet, so behavior is unchanged; the splice entry points that persist them follow. Generated with assistance from Claude Code. Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Fable 5 --- src/payment/pending_payment_store.rs | 8 ++ src/wallet/mod.rs | 115 +++++++++++++++++++-------- 2 files changed, 92 insertions(+), 31 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 05772d259b..3cfefe995d 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -169,6 +169,14 @@ impl PendingPaymentDetails { } } + /// The splice intent this record carries, if it is a splice that has not yet locked. + pub(crate) fn splice_intent(&self) -> Option<&SpliceIntent> { + match self { + Self::PendingSplice { intent, .. } => Some(intent), + Self::Tracked { splice_intent, .. } => splice_intent.as_ref(), + } + } + /// Returns this node's recorded funding figures for the candidate with the given txid, if any. pub(crate) fn candidate(&self, txid: Txid) -> Option<&FundingTxCandidate> { match self { diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index b309436f75..8ec1076f96 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1266,6 +1266,24 @@ impl Wallet { Ok(()) } + /// Returns the `PaymentId` of a user-initiated splice intent for one of the channels in + /// `candidate`, if any, so a classified splice adopts the id chosen at splice time rather than + /// deriving one from the first candidate's txid. A fee bump reuses the channel's existing intent, + /// so at most one in-flight intent matches and the first is unambiguous. + fn find_splice_payment_id(&self, candidate: &FundingCandidate) -> Option { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|intent| { + candidate.channels.iter().any(|channel| { + channel.channel_id == intent.channel_id + && channel.counterparty_node_id == intent.counterparty_node_id + }) + }) + }) + .first() + .map(|p| p.id()) + } + /// Records an interactive-funding broadcast (splice, or a V2 dual-funded open) as a pending /// on-chain payment, tagged with its transaction type. Amount and fee are this node's share, /// derived from the active candidate's contributions; broadcasts we didn't contribute to, or @@ -1316,9 +1334,18 @@ impl Wallet { return Ok(()); } - // Anchor the `PaymentId` to the first negotiated candidate so the record stays stable - // across RBF replacements. - let payment_id = PaymentId(first.txid.to_byte_array()); + // Adopt the `PaymentId` generated when the splice was initiated so its retry intent, funding + // payment, and candidate history share one record. If the intent is already gone (e.g. the + // splice locked before this classification ran), adopt the id of a record wallet sync + // created for any candidate rather than minting a divergent one. Fall back to the first + // negotiated candidate's txid for splices we did not originate (counterparty-initiated or + // V2 opens), which keeps that id stable across RBF replacements. + let payment_id = self + .find_splice_payment_id(active) + .or_else(|| { + candidates.iter().find_map(|candidate| self.find_payment_by_txid(candidate.txid)) + }) + .unwrap_or_else(|| PaymentId(first.txid.to_byte_array())); // Record every candidate's figures (`None` for any round we didn't contribute to, e.g. a // counterparty-initiated splice our `splice_in` later joined via RBF) so the confirmed @@ -1411,35 +1438,57 @@ impl Wallet { &candidates, self.payment_store.get(&details.id).as_ref(), ); - let pending_update = PendingPaymentDetailsUpdate { - id: update.id, - payment_update: Some(update.clone()), - conflicting_txids: None, - candidates: candidates.clone(), - splice_intent: None, - }; - self.payment_store.update_or_insert(update, details.clone()).await?; - - // The pending index must exist exactly while the authoritative record is Pending: - // graduation and rebroadcast read it, and a graduated payment must not be re-indexed. - // Deciding by the post-write status rather than by whether the write inserted also - // repairs a missing index — a crash or failed write between the two stores leaves a - // Pending record with no entry, and a merge alone would never recreate it, leaving the - // payment unable to graduate and its txids unmapped. + self.payment_store.update_or_insert(update.clone(), details.clone()).await?; + + // The post-write record decides what the pending index may hold: only `Pending` payments + // belong in the pending store (graduation and rebroadcast assume it), and a promoted or + // (re)created entry must carry the record wallet sync advanced, not the fresh + // Unconfirmed details. let recorded = self.payment_store.get(&details.id).unwrap_or(details); - if recorded.status == PaymentStatus::Pending { - // Wallet sync can still land between the payment-store write above and this one and - // mirror an advanced confirmation into the pending store, so this write makes the - // same atomic decision: merge narrowly into an entry that appeared, insert otherwise. - // The inserted entry embeds the post-write record rather than the fresh details, so a - // confirmation wallet sync already recorded keeps driving graduation. - let pending = PendingPaymentDetails::new(recorded, Vec::new(), candidates); - self.pending_payment_store.update_or_insert(pending_update, pending).await?; - } else { - // The payment already advanced beyond Pending: the graduation path removed the - // entry, and `update`'s no-op on absence must not re-create it. - self.pending_payment_store.update(pending_update).await?; - } + let id = recorded.id; + self.pending_payment_store + .mutate(&id, |existing| match existing { + // First time we record this funding payment — or a crash between the two store + // writes left a Pending record with no index entry: (re)create it so the payment + // can graduate and its candidate txids stay mapped. A graduated payment is never + // `Pending`, so absence with an advanced record means the graduation path removed + // the entry and it must not be re-indexed. + None => (recorded.status == PaymentStatus::Pending).then(|| { + PendingPaymentDetails::tracked(recorded, Vec::new(), candidates, None) + }), + // A user-initiated splice has a pre-broadcast `PendingSplice` intent under this + // id; carry its intent into the `Tracked` record so the retrier can still clear + // it once the splice locks. If the payment already advanced beyond `Pending` + // (wallet sync confirmed it through `ANTI_REORG_DELAY` first), it must not enter + // the pending store; the intent stays for `ChannelReady` or `reconcile` to clear. + Some(PendingPaymentDetails::PendingSplice { intent, .. }) => { + if recorded.status == PaymentStatus::Pending { + Some(PendingPaymentDetails::tracked( + recorded, + Vec::new(), + candidates, + Some(intent.clone()), + )) + } else { + None + } + }, + // An earlier candidate or a racing wallet sync already recorded this payment: + // merge only the classification (`tx_type`, candidate history and the figures of + // whichever candidate the record's state makes authoritative) into it. + Some(tracked @ PendingPaymentDetails::Tracked { .. }) => { + let mut updated = tracked.clone(); + let pending_update = PendingPaymentDetailsUpdate { + id, + payment_update: Some(update), + conflicting_txids: None, + candidates, + splice_intent: None, + }; + updated.update(pending_update).then_some(updated) + }, + }) + .await?; Ok(()) } @@ -1549,6 +1598,10 @@ impl Wallet { p.details().is_some_and( |d| matches!(d.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid), ) || p.conflicting_txids().contains(&target_txid) + // A splice keyed by a generated PaymentId is not found by the txid-derived id + // above, so map any of its candidate txids (an earlier RBF round may confirm) + // back to the record. + || p.candidate(target_txid).is_some() }) .first() { From 1653bedef3143f92ba2571458dfc97da78cb241e Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 2 Jul 2026 00:29:02 -0500 Subject: [PATCH 08/11] Persist and resubmit user-initiated splices across restarts LDK abandons an in-progress splice negotiation whenever the peer disconnects -- which includes stopping the node -- and only durably records a splice once its negotiation reaches signing. A splice dropped before then, after splice_in, splice_out, or bump_channel_funding_fee returned Ok, is therefore silently lost across a restart or an ill-timed disconnect. Persist a splice intent before handing the contribution to LDK, keyed by a PaymentId generated at splice time and reusing the channel's existing intent record when one is present, so a splice and its fee bumps share one id and at most one intent exists per channel. At startup a reconciler probes each intent against LDK's live channel state and resubmits any LDK dropped -- including those lost to a crash before LDK persisted anything -- surfacing SpliceNegotiationFailed only when the channel is gone, a fee bump has nothing left to replace, or the resubmission budget is exhausted. Resubmitting does not require the peer to be connected: LDK holds the contribution and initiates quiescence on reconnect. Wallet sync can see the splice transaction before the broadcast-time classification records it -- the counterparty broadcasts it too -- but a pre-broadcast intent record carries no txids for the usual lookup to match. Teach sync to recognize such a transaction by the funding outpoint it spends and adopt the splice-time id, so both writers converge on one record instead of sync minting a txid-derived duplicate. A payment-tracking merge (e.g. from wallet sync) must leave a live intent untouched. The splice tests now locate a funding payment by its candidate txid, since its id is generated rather than derived from the funding txid. Generated with assistance from Claude Code. Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Fable 5 --- src/builder.rs | 1 + src/channel/mod.rs | 202 +++++++++++++++++++++++++++ src/lib.rs | 143 ++++++++++++++++++- src/payment/pending_payment_store.rs | 15 +- src/wallet/mod.rs | 38 ++++- tests/integration_tests_rust.rs | 34 +++-- 6 files changed, 406 insertions(+), 27 deletions(-) create mode 100644 src/channel/mod.rs diff --git a/src/builder.rs b/src/builder.rs index f117800996..685278a795 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -2333,6 +2333,7 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + pending_payment_store, lnurl_auth, is_running, node_metrics, diff --git a/src/channel/mod.rs b/src/channel/mod.rs new file mode 100644 index 0000000000..bc48581caa --- /dev/null +++ b/src/channel/mod.rs @@ -0,0 +1,202 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Retrying user-initiated splices that LDK dropped before durably recording them. + +use std::ops::Deref; +use std::sync::Arc; + +use bitcoin::secp256k1::PublicKey; +use lightning::ln::channelmanager::PaymentId; +use lightning::ln::types::ChannelId; + +use crate::data_store::StorableObject; +use crate::event::{Event, EventQueue}; +use crate::logger::{log_error, log_info, LdkLogger}; +use crate::payment::pending_payment_store::{ + PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, MAX_SPLICE_ATTEMPTS, +}; +use crate::types::{ChannelManager, PendingPaymentStore}; +use crate::Error; + +/// Resubmits user-initiated splices that LDK dropped before durably recording them. +/// +/// LDK only persists a splice once its negotiation reaches `AwaitingSignatures`, and it abandons an +/// earlier negotiation whenever the peer disconnects (which includes restarting the node). The +/// splice entry points persist a [`SpliceIntent`] before handing the contribution to LDK; this type +/// drives that intent back into [`ChannelManager::funding_contributed`] until the splice either +/// locks (clearing the intent) or fails for a reason retrying cannot address. +/// +/// Resubmitting does not require the peer to be connected: LDK holds on to the contribution and +/// initiates quiescence once the peer reconnects. +/// +/// [`ChannelManager::funding_contributed`]: lightning::ln::channelmanager::ChannelManager::funding_contributed +pub(crate) struct SpliceRetrier +where + L::Target: LdkLogger, +{ + channel_manager: Arc, + pending_payment_store: Arc, + event_queue: Arc>, + logger: L, +} + +impl SpliceRetrier +where + L::Target: LdkLogger, +{ + pub(crate) fn new( + channel_manager: Arc, pending_payment_store: Arc, + event_queue: Arc>, logger: L, + ) -> Self { + Self { channel_manager, pending_payment_store, event_queue, logger } + } + + /// Reconciles persisted splice intents against live channel state. Run once at startup to pick + /// up splices LDK dropped before durably recording them — including those lost to a crash before + /// LDK persisted anything. + pub(crate) async fn reconcile(&self) { + let records = self.pending_payment_store.list_filter(|p| p.splice_intent().is_some()); + for record in records { + let id = record.id(); + let has_payment = record.details().is_some(); + let Some(intent) = record.splice_intent().cloned() else { + continue; + }; + + let channel = self + .channel_manager + .list_channels_with_counterparty(&intent.counterparty_node_id) + .into_iter() + .find(|c| c.user_channel_id == intent.user_channel_id.0); + let channel = match channel { + Some(channel) => channel, + None => { + // The channel is gone; there is nothing to splice anymore. + self.clear_intent(id, has_payment).await; + continue; + }, + }; + + if channel.funding_txo != Some(intent.pre_splice_funding_txo) { + // The funding moved on, so the splice (or a replacement) locked. + self.clear_intent(id, has_payment).await; + continue; + } + + // `splice_channel` is a read-only probe of LDK's splice state. It fails when we already + // have a splice in flight (a held contribution, an in-progress negotiation, or one + // awaiting signatures), all of which LDK drives to completion on its own. + let template = match self + .channel_manager + .splice_channel(&channel.channel_id, &intent.counterparty_node_id) + { + Ok(template) => template, + Err(_) => continue, + }; + + // LDK persists a splice once negotiated, so a prior contribution means the intent was + // carried out — unless the intent was a fee bump at a higher feerate than negotiated. + let should_retry = match (&intent.kind, template.prior_contribution()) { + (SpliceKind::Rbf {}, Some(prior)) => { + prior.feerate() < intent.contribution.feerate() + }, + (SpliceKind::Rbf {}, None) => { + // The splice to bump is gone entirely; surface rather than guess. + self.abandon(id, has_payment, &intent).await; + continue; + }, + (_, Some(_)) => false, + (_, None) => true, + }; + if !should_retry { + continue; + } + + if intent.attempts >= MAX_SPLICE_ATTEMPTS { + self.abandon(id, has_payment, &intent).await; + continue; + } + + log_info!( + self.logger, + "Resubmitting splice for channel {} with counterparty {}", + channel.channel_id, + intent.counterparty_node_id, + ); + let counterparty_node_id = intent.counterparty_node_id; + let _ = self.submit(id, &channel.channel_id, &counterparty_node_id, intent).await; + } + } + + /// Persists the incremented attempt count and hands the contribution back to LDK. The count is + /// persisted first so that a crash mid-submission cannot lead to unbounded retries. + async fn submit( + &self, id: PaymentId, channel_id: &ChannelId, counterparty_node_id: &PublicKey, + mut intent: SpliceIntent, + ) -> Result<(), Error> { + intent.attempts += 1; + let contribution = intent.contribution.clone(); + let update = PendingPaymentDetailsUpdate { + id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent)), + }; + self.pending_payment_store.update(update).await?; + + self.channel_manager + .funding_contributed(channel_id, counterparty_node_id, contribution, None) + .map_err(|e| { + log_error!( + self.logger, + "Failed to resubmit splice for channel {} with counterparty {}: {:?}", + channel_id, + counterparty_node_id, + e, + ); + Error::ChannelSplicingFailed + }) + } + + /// Drops a splice intent: removes a pre-broadcast record entirely, or clears just the intent on + /// a record that already carries a classified funding payment so the payment keeps graduating. + async fn clear_intent(&self, id: PaymentId, has_payment: bool) { + if has_payment { + let update = PendingPaymentDetailsUpdate { + id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(None), + }; + let _ = self.pending_payment_store.update(update).await; + } else { + let _ = self.pending_payment_store.remove(&id).await; + } + } + + /// Gives up on a splice intent and surfaces the failure to the user. + async fn abandon(&self, id: PaymentId, has_payment: bool, intent: &SpliceIntent) { + log_error!( + self.logger, + "Abandoning splice for channel {} with counterparty {}", + intent.channel_id, + intent.counterparty_node_id, + ); + self.clear_intent(id, has_payment).await; + let event = Event::SpliceNegotiationFailed { + channel_id: intent.channel_id, + user_channel_id: intent.user_channel_id, + counterparty_node_id: intent.counterparty_node_id, + }; + if let Err(e) = self.event_queue.add_event(event).await { + log_error!(self.logger, "Failed to push to event queue: {}", e); + } + } +} diff --git a/src/lib.rs b/src/lib.rs index b22c1538cc..a14bcd42dd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -83,6 +83,7 @@ mod balance; mod builder; mod chain; +mod channel; pub mod config; mod connection; mod data_store; @@ -128,12 +129,14 @@ pub use builder::BuildError; #[cfg(not(feature = "uniffi"))] pub use builder::NodeBuilder as Builder; use chain::ChainSource; +use channel::SpliceRetrier; use config::{ default_user_config, may_announce_channel, AsyncPaymentsRole, ChannelConfig, Config, LNURL_AUTH_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL, PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL, }; use connection::ConnectionManager; +use data_store::StorableObject; pub use error::Error as NodeError; use error::Error; pub use event::Event; @@ -153,6 +156,7 @@ use lightning::ln::chan_utils::FUNDING_TRANSACTION_WITNESS_WEIGHT; use lightning::ln::channel_state::ChannelDetails as LdkChannelDetails; pub use lightning::ln::channel_state::ChannelShutdownState; use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; use lightning::ln::msgs::{BaseMessageHandler, SocketAddress}; use lightning::ln::peer_handler::CustomMessageHandler; use lightning::routing::gossip::NodeAlias; @@ -171,6 +175,9 @@ use lnurl_auth::LnurlAuth; use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use payment::asynchronous::om_mailbox::OnionMessageMailbox; use payment::asynchronous::static_invoice_store::StaticInvoiceStore; +use payment::pending_payment_store::{ + PendingPaymentDetails, PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, +}; use payment::{ Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment, UnifiedPayment, @@ -183,8 +190,8 @@ use runtime::Runtime; pub use tokio; use types::{ Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph, - HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper, - Wallet, + HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, + Router, Scorer, Sweeper, Wallet, }; pub use types::{ ChannelCounterparty, ChannelDetails, CustomTlvRecord, PeerDetails, ReserveType, UserChannelId, @@ -265,6 +272,7 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + pending_payment_store: Arc, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -651,6 +659,13 @@ impl Node { None }; + let splice_retrier = Arc::new(SpliceRetrier::new( + Arc::clone(&self.channel_manager), + Arc::clone(&self.pending_payment_store), + Arc::clone(&self.event_queue), + Arc::clone(&self.logger), + )); + let event_handler = Arc::new(EventHandler::new( Arc::clone(&self.event_queue), Arc::clone(&self.wallet), @@ -679,6 +694,11 @@ impl Node { }); } + // Resubmit any persisted splice intents that LDK dropped before durably recording them. + self.runtime.spawn_background_task(async move { + splice_retrier.reconcile().await; + }); + // Setup background processing let background_persister = Arc::clone(&self.kv_store); let background_event_handler = Arc::clone(&event_handler); @@ -1655,6 +1675,87 @@ impl Node { ) } + /// Persists a splice intent before its contribution is handed to LDK, so the splice can be + /// resubmitted if LDK drops it before durably recording it (a restart, or a disconnect + /// mid-negotiation). Must be called before `funding_contributed` so a crash in between is also + /// covered. + /// + /// Reuses the channel's existing splice intent record when one is present -- so a splice and its + /// later fee bumps share one [`PaymentId`] and at most one intent ever exists per channel, which + /// [`Wallet::find_splice_payment_id`] and the retrier rely on -- otherwise generates a fresh id. + /// Returns the id and, for restoring on failure, `None` when a fresh record was created or + /// `Some(prior)` when an existing record's intent was replaced. + fn persist_splice_intent( + &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, + channel_details: &LdkChannelDetails, contribution: FundingContribution, kind: SpliceKind, + ) -> Result<(PaymentId, Option>), Error> { + let pre_splice_funding_txo = channel_details.funding_txo.ok_or_else(|| { + log_error!(self.logger, "Failed to splice channel: channel not yet ready"); + Error::ChannelSplicingFailed + })?; + let intent = SpliceIntent { + user_channel_id: *user_channel_id, + counterparty_node_id, + channel_id: channel_details.channel_id, + pre_splice_funding_txo, + contribution, + kind, + attempts: 0, + }; + let existing = self + .pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|i| i.user_channel_id == *user_channel_id) + }) + .into_iter() + .next(); + match existing { + Some(record) => { + let payment_id = record.id(); + let prior = record.splice_intent().cloned(); + self.runtime.block_on(self.pending_payment_store.update( + PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent)), + }, + ))?; + Ok((payment_id, Some(prior))) + }, + None => { + let payment_id = PaymentId(self.keys_manager.get_secure_random_bytes()); + self.runtime.block_on( + self.pending_payment_store + .insert(PendingPaymentDetails::pending_splice(payment_id, intent)), + )?; + Ok((payment_id, None)) + }, + } + } + + /// Undoes a splice intent persisted for an originating call whose `funding_contributed` then + /// failed: restores an existing record's prior intent, or removes a freshly created record. + fn discard_splice_intent(&self, payment_id: &PaymentId, restore: Option>) { + match restore { + Some(prior) => { + let _ = self.runtime.block_on(self.pending_payment_store.update( + PendingPaymentDetailsUpdate { + id: *payment_id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(prior), + }, + )); + }, + None => { + let _ = self.runtime.block_on(self.pending_payment_store.remove(payment_id)); + }, + } + } + fn splice_in_inner( &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, splice_amount_sats: FundingAmount, @@ -1763,6 +1864,14 @@ impl Node { Error::ChannelSplicingFailed })?; + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::In { amount_sats: splice_amount_sats }, + )?; + self.channel_manager .funding_contributed( &channel_details.channel_id, @@ -1772,6 +1881,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to splice channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { @@ -1887,11 +1997,20 @@ impl Node { value: Amount::from_sat(splice_amount_sats), script_pubkey: address.script_pubkey(), }]; - let contribution = - funding_template.splice_out(outputs, feerate, max_feerate).map_err(|e| { - log_error!(self.logger, "Failed to splice channel: {}", e); - Error::ChannelSplicingFailed - })?; + let contribution = funding_template + .splice_out(outputs.clone(), feerate, max_feerate) + .map_err(|e| { + log_error!(self.logger, "Failed to splice channel: {}", e); + Error::ChannelSplicingFailed + })?; + + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::Out { outputs }, + )?; self.channel_manager .funding_contributed( @@ -1902,6 +2021,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to splice channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { @@ -1964,6 +2084,14 @@ impl Node { Error::ChannelSplicingFailed })?; + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::Rbf {}, + )?; + self.channel_manager .funding_contributed( &channel_details.channel_id, @@ -1973,6 +2101,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to RBF channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 3cfefe995d..43f022d37f 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -18,6 +18,10 @@ use crate::payment::store::PaymentDetailsUpdate; use crate::payment::{PaymentDetails, PaymentKind}; use crate::types::UserChannelId; +/// The number of times a splice intent is resubmitted to LDK before it is abandoned and the +/// failure is surfaced to the user. +pub(crate) const MAX_SPLICE_ATTEMPTS: u8 = 3; + /// One candidate transaction in an interactive-funding (splice) RBF history, holding this node's /// share of the funding amount and fee for that candidate. Both are `None` for a candidate this /// node did not contribute to — e.g. a counterparty-initiated round before our `splice_in` joined @@ -153,6 +157,10 @@ impl PendingPaymentDetails { Self::Tracked { details, conflicting_txids, candidates, splice_intent } } + pub(crate) fn pending_splice(id: PaymentId, intent: SpliceIntent) -> Self { + Self::PendingSplice { id, intent } + } + /// The full payment details, or `None` for a splice not yet broadcast. pub(crate) fn details(&self) -> Option<&PaymentDetails> { match self { @@ -311,12 +319,17 @@ impl From<&PendingPaymentDetails> for PendingPaymentDetailsUpdate { } else { Some(conflicting_txids.clone()) }; + // Leave the splice intent unchanged: it is owned by the splice entry points and the + // retrier, never by a payment-tracking merge. Emitting the current value here would + // let an `insert_or_update` of a payment record (e.g. from wallet sync, built without + // an intent) clobber a live intent to `None`. + let _ = splice_intent; Self { id: details.id, payment_update: Some(details.to_update()), conflicting_txids, candidates: candidates.clone(), - splice_intent: Some(splice_intent.clone()), + splice_intent: None, } }, } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 8ec1076f96..b64e227323 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -57,7 +57,7 @@ use crate::config::Config; use crate::data_store::StorableObject; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; -use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; +use crate::payment::pending_payment_store::{PendingPaymentDetailsUpdate, SpliceKind}; use crate::payment::store::{ConfirmationStatus, PaymentDetailsUpdate}; use crate::payment::{ FundingTxCandidate, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, @@ -258,7 +258,7 @@ impl Wallet { }; let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -360,7 +360,7 @@ impl Wallet { }, WalletEvent::TxUnconfirmed { txid, tx, .. } => { let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -417,7 +417,7 @@ impl Wallet { }, WalletEvent::TxDropped { txid, tx } => { let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -1611,6 +1611,36 @@ impl Wallet { None } + /// Like [`Self::find_payment_by_txid`], but additionally recognizes the transaction of a + /// user-initiated splice that has no payment record yet: it spends the funding outpoint a + /// live splice intent was created for. Wallet sync can see the transaction before the + /// broadcast-time classification records it — the counterparty broadcasts it too — and must + /// adopt the splice-time `PaymentId` so both writers converge on one record. + fn find_payment_for_tx(&self, tx: &Transaction, txid: Txid) -> Option { + self.find_payment_by_txid(txid).or_else(|| { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|intent| { + // A cooperative or force close also spends the funding outpoint, and a + // cooperative close pays a wallet address. A splice-in always adds wallet + // inputs while a close never has more than the funding input, so a second + // input disambiguates `SpliceKind::In`. A splice-out or fee bump can share + // the close's single-input shape; misattributing a close that races a + // still-live intent only affects which id keys its record. + let input_shape_matches = match intent.kind { + SpliceKind::In { .. } => tx.input.len() > 1, + SpliceKind::Out { .. } | SpliceKind::Rbf {} => true, + }; + let funding_txo = intent.pre_splice_funding_txo.into_bitcoin_outpoint(); + input_shape_matches + && tx.input.iter().any(|input| input.previous_output == funding_txo) + }) + }) + .first() + .map(|p| p.id()) + }) + } + /// 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 diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index e401c82189..62f9615b28 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -216,6 +216,18 @@ fn wallet_store_contention_does_not_stall_runtime() { result.unwrap_or_else(|e| panic!("wallet contention test failed: {e}")); } +/// Finds an on-chain funding payment by its active candidate `txid`. A user-initiated splice's +/// `PaymentId` is generated at splice time rather than derived from a txid, so the payment must be +/// located by `kind.txid` (the active or confirmed candidate) instead of a txid-derived id. +fn funding_payment(node: &Node, txid: Txid) -> PaymentDetails { + node.list_payments_with_filter( + |p| matches!(p.kind, PaymentKind::Onchain { txid: candidate, .. } if candidate == txid), + ) + .into_iter() + .next() + .expect("no funding payment for the given txid") +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn channel_full_cycle() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -1776,9 +1788,7 @@ async fn splice_channel() { // them to the channel balance since there may not be a change output. let expected_splice_in_lightning_balance_sat = 4_000_002; - let payments = node_b.list_payments(); - let payment = - payments.into_iter().find(|p| p.id == PaymentId(txo.txid.to_byte_array())).unwrap(); + let payment = funding_payment(&node_b, txo.txid); assert_eq!(payment.fee_paid_msat, Some(expected_splice_in_fee_sat * 1_000)); assert_eq!( @@ -1829,9 +1839,7 @@ async fn splice_channel() { let expected_splice_out_fee_sat = 183; - let payments = node_a.list_payments(); - let payment = - payments.into_iter().find(|p| p.id == PaymentId(txo.txid.to_byte_array())).unwrap(); + let payment = funding_payment(&node_a, txo.txid); assert_eq!(payment.fee_paid_msat, Some(expected_splice_out_fee_sat * 1_000)); // The splice-out graduated to a confirmed interactive-funding payment. Its `direction` is left // unasserted on purpose: the destination is our own address, so it is a self-transfer (channel @@ -1943,8 +1951,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // For `confirm_original`, capture the original candidate's fee and raw transaction now, before // the RBF replaces it, so it can be force-confirmed (instead of the RBF) further below. let original_candidate: Option<(Option, String)> = if confirm_original { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let fee = node_b.payment(&payment_id).expect("splice payment exists").fee_paid_msat; + let fee = funding_payment(&node_b, original_txo.txid).fee_paid_msat; let raw_tx: String = bitcoind .client .call("getrawtransaction", &[json!(original_txo.txid.to_string())]) @@ -1982,8 +1989,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // at the latest (RBF) candidate, and the durable interactive-funding `tx_type` preserved across // the replacement. let rbf_candidate_fee = { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment exists"); + let payment = funding_payment(&node_b, rbf_txo.txid); match payment.kind { PaymentKind::Onchain { txid, @@ -2057,8 +2063,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // channel-lifecycle signal, not what drives payment status. Its `kind.txid` reflects the // winning RBF candidate, and `fee_paid_msat` carries this node's `FundingContribution` fee. { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment graduated"); + let payment = funding_payment(&node_b, winning_txo.txid); assert_eq!(payment.status, PaymentStatus::Succeeded); match payment.kind { PaymentKind::Onchain { txid, status: ConfirmationStatus::Confirmed { .. }, .. } => { @@ -2182,8 +2187,7 @@ async fn splice_payment_reorged_to_unconfirmed() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; node_b.sync_wallets().unwrap(); - let payment_id = PaymentId(splice_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment exists"); + let payment = funding_payment(&node_b, splice_txo.txid); assert_eq!(payment.status, PaymentStatus::Pending); assert!(matches!( payment.kind, @@ -2206,7 +2210,7 @@ async fn splice_payment_reorged_to_unconfirmed() { // The funding payment returns to `Unconfirmed` and stays `Pending`, exercising the // `TxUnconfirmed` arm for a funding payment. - let payment = node_b.payment(&payment_id).expect("splice payment still exists"); + let payment = funding_payment(&node_b, splice_txo.txid); assert_eq!(payment.status, PaymentStatus::Pending); assert!(matches!( payment.kind, From d3bbdc8f1da9900a943ccd9f7c121e783b60da6b Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 2 Jul 2026 00:39:08 -0500 Subject: [PATCH 09/11] Retry recoverable splice failures and emit only final give-up A user-initiated splice can fail mid-negotiation while the node is running -- the peer disconnects, or the contribution goes stale behind a competing negotiation -- and LDK reports each such round via SpliceNegotiationFailed. Drive those events through the splice retrier: resubmit the same contribution when the peer merely disconnected, rebuild a fresh one when it went stale, and give up (surfacing the failure) only for a non-retriable reason or once the resubmission budget is exhausted, using LDK's own is_retriable classification. Clear a splice's intent once the channel locks its new funding or the channel closes. Event::SpliceNegotiationFailed is now emitted only when a splice is finally abandoned, not for every failed negotiation round, since a recoverable failure is retried transparently. Generated with assistance from Claude Code. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/channel/mod.rs | 229 ++++++++++++++++++++++++++++++++++++++++++++- src/event.rs | 34 ++++++- src/lib.rs | 3 + 3 files changed, 258 insertions(+), 8 deletions(-) diff --git a/src/channel/mod.rs b/src/channel/mod.rs index bc48581caa..09c03f0a39 100644 --- a/src/channel/mod.rs +++ b/src/channel/mod.rs @@ -11,18 +11,54 @@ use std::ops::Deref; use std::sync::Arc; use bitcoin::secp256k1::PublicKey; +use bitcoin::{Amount, OutPoint}; +use lightning::events::NegotiationFailureReason; use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; use lightning::ln::types::ChannelId; use crate::data_store::StorableObject; use crate::event::{Event, EventQueue}; +use crate::fee_estimator::{ + max_funding_feerate, ConfirmationTarget, FeeEstimator, OnchainFeeEstimator, +}; use crate::logger::{log_error, log_info, LdkLogger}; use crate::payment::pending_payment_store::{ - PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, MAX_SPLICE_ATTEMPTS, + PendingPaymentDetails, PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, + MAX_SPLICE_ATTEMPTS, }; -use crate::types::{ChannelManager, PendingPaymentStore}; +use crate::types::{ChannelManager, PendingPaymentStore, UserChannelId, Wallet}; use crate::Error; +/// The action to take on a `SpliceNegotiationFailed` for a splice intent we track, decided purely +/// from the failure `reason` and the intent's attempt count so the decision matrix can be +/// unit-tested without a live channel. A failure for a splice we don't track is surfaced directly +/// (see [`SpliceRetrier::on_negotiation_failed`]) and never reaches here. +#[derive(Debug, PartialEq, Eq)] +enum RetryDecision { + /// Give up: clear the intent and surface the failure to the user. + Abandon, + /// Resubmit the stored contribution unchanged (a transient failure such as a disconnect). + ResubmitStored, + /// Rebuild a fresh contribution from the original parameters (the stored one went stale). + Rebuild, +} + +fn decide_retry(reason: &NegotiationFailureReason, attempts: u8) -> RetryDecision { + if !reason.is_retriable() || attempts >= MAX_SPLICE_ATTEMPTS { + return RetryDecision::Abandon; + } + match reason { + // The stored contribution is still valid after a transient failure. + NegotiationFailureReason::PeerDisconnected | NegotiationFailureReason::Unknown => { + RetryDecision::ResubmitStored + }, + // The remaining retriable reasons (`FeeRateTooLow`, `ContributionInvalid`) mean the stored + // contribution went stale. + _ => RetryDecision::Rebuild, + } +} + /// Resubmits user-initiated splices that LDK dropped before durably recording them. /// /// LDK only persists a splice once its negotiation reaches `AwaitingSignatures`, and it abandons an @@ -40,6 +76,8 @@ where L::Target: LdkLogger, { channel_manager: Arc, + wallet: Arc, + fee_estimator: Arc, pending_payment_store: Arc, event_queue: Arc>, logger: L, @@ -50,10 +88,11 @@ where L::Target: LdkLogger, { pub(crate) fn new( - channel_manager: Arc, pending_payment_store: Arc, + channel_manager: Arc, wallet: Arc, + fee_estimator: Arc, pending_payment_store: Arc, event_queue: Arc>, logger: L, ) -> Self { - Self { channel_manager, pending_payment_store, event_queue, logger } + Self { channel_manager, wallet, fee_estimator, pending_payment_store, event_queue, logger } } /// Reconciles persisted splice intents against live channel state. Run once at startup to pick @@ -199,4 +238,186 @@ where log_error!(self.logger, "Failed to push to event queue: {}", e); } } + + /// Applies a `SpliceNegotiationFailed` to any matching splice intent, retrying recoverable + /// failures. Returns whether the failure should be surfaced to the user (i.e. the splice is + /// given up on). + pub(crate) async fn on_negotiation_failed( + &self, user_channel_id: UserChannelId, reason: NegotiationFailureReason, + contribution: Option, + ) -> bool { + let Some(record) = self.record_for_channel(user_channel_id) else { + return true; + }; + let id = record.id(); + let has_payment = record.details().is_some(); + let Some(intent) = record.splice_intent().cloned() else { + return true; + }; + + // Only act on failures of the splice we are tracking. A mismatch means the failure concerns + // some other attempt (e.g. a stale event replayed after a newer splice was initiated). + if contribution.as_ref() != Some(&intent.contribution) { + return true; + } + + let channel_id = intent.channel_id; + let counterparty_node_id = intent.counterparty_node_id; + match decide_retry(&reason, intent.attempts) { + RetryDecision::Abandon => { + self.clear_intent(id, has_payment).await; + true + }, + RetryDecision::ResubmitStored => { + // The same contribution remains valid; resubmit it. Skip if LDK already has a splice + // in flight for this channel (e.g. the startup reconciler resubmitted first). + if self.channel_manager.splice_channel(&channel_id, &counterparty_node_id).is_err() + { + return false; + } + log_info!( + self.logger, + "Resubmitting splice for channel {} with counterparty {} after a recoverable failure", + channel_id, + counterparty_node_id, + ); + let _ = self.submit(id, &channel_id, &counterparty_node_id, intent).await; + false + }, + RetryDecision::Rebuild => { + // The stored contribution went stale; rebuild a fresh one from the original params. + match self + .rebuild_contribution(&channel_id, &counterparty_node_id, &intent.kind) + .await + { + Ok(contribution) => { + log_info!( + self.logger, + "Resubmitting rebuilt splice for channel {} with counterparty {}", + channel_id, + counterparty_node_id, + ); + let mut intent = intent; + intent.contribution = contribution; + let _ = self.submit(id, &channel_id, &counterparty_node_id, intent).await; + false + }, + Err(e) => { + log_error!( + self.logger, + "Abandoning splice for channel {}: failed to rebuild contribution: {:?}", + channel_id, + e, + ); + self.clear_intent(id, has_payment).await; + true + }, + } + }, + } + } + + /// Clears any splice intent made obsolete by a newly locked funding transaction. + pub(crate) async fn on_channel_ready( + &self, user_channel_id: UserChannelId, funding_txo: Option, + ) { + let Some(record) = self.record_for_channel(user_channel_id) else { + return; + }; + let id = record.id(); + let has_payment = record.details().is_some(); + let Some(intent) = record.splice_intent() else { + return; + }; + // Only clear an intent that predates the locked funding. An intent whose pre-splice outpoint + // still matches the newly locked funding was created after this lock and is still pending. + let clear = match funding_txo { + Some(funding_txo) => { + intent.pre_splice_funding_txo.into_bitcoin_outpoint() != funding_txo + }, + None => false, + }; + if clear { + self.clear_intent(id, has_payment).await; + } + } + + /// Clears any splice intent for a closed channel, as there is nothing left to splice. + pub(crate) async fn on_channel_closed(&self, user_channel_id: UserChannelId) { + if let Some(record) = self.record_for_channel(user_channel_id) { + self.clear_intent(record.id(), record.details().is_some()).await; + } + } + + /// Returns the pending record carrying a splice intent for the given channel, if any. + fn record_for_channel(&self, user_channel_id: UserChannelId) -> Option { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|i| i.user_channel_id == user_channel_id) + }) + .into_iter() + .next() + } + + /// Builds a fresh contribution from the parameters of the originating API call, mirroring the + /// corresponding [`Node`] method. + /// + /// [`Node`]: crate::Node + async fn rebuild_contribution( + &self, channel_id: &ChannelId, counterparty_node_id: &PublicKey, kind: &SpliceKind, + ) -> Result { + let template = self + .channel_manager + .splice_channel(channel_id, counterparty_node_id) + .map_err(|_| Error::ChannelSplicingFailed)?; + + let est_feerate = self.fee_estimator.estimate_fee_rate(ConfirmationTarget::ChannelFunding); + let max_feerate = max_funding_feerate(est_feerate); + let feerate = match template.min_rbf_feerate() { + Some(min_rbf_feerate) if min_rbf_feerate <= max_feerate => { + est_feerate.max(min_rbf_feerate) + }, + _ => est_feerate, + }; + + match kind { + SpliceKind::In { amount_sats } => template + .splice_in( + Amount::from_sat(*amount_sats), + feerate, + max_feerate, + Arc::clone(&self.wallet), + ) + .await + .map_err(|_| Error::ChannelSplicingFailed), + SpliceKind::Out { outputs } => template + .splice_out(outputs.clone(), feerate, max_feerate) + .map_err(|_| Error::ChannelSplicingFailed), + SpliceKind::Rbf {} => template + .rbf_prior_contribution(None, max_feerate, Arc::clone(&self.wallet)) + .await + .map_err(|_| Error::ChannelSplicingFailed), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn decide_retry_matrix() { + use NegotiationFailureReason::*; + + // A non-retriable reason gives up regardless of attempts. + assert_eq!(decide_retry(&LocallyCanceled, 0), RetryDecision::Abandon); + // Retriable, but the resubmission budget is exhausted -> give up. + assert_eq!(decide_retry(&PeerDisconnected, MAX_SPLICE_ATTEMPTS), RetryDecision::Abandon); + // Transient failures resubmit the stored contribution. + assert_eq!(decide_retry(&PeerDisconnected, 0), RetryDecision::ResubmitStored); + assert_eq!(decide_retry(&Unknown, MAX_SPLICE_ATTEMPTS - 1), RetryDecision::ResubmitStored); + // A stale contribution is rebuilt from the original parameters. + assert_eq!(decide_retry(&FeeRateTooLow, 0), RetryDecision::Rebuild); + assert_eq!(decide_retry(&ContributionInvalid, 0), RetryDecision::Rebuild); + } } diff --git a/src/event.rs b/src/event.rs index b8ca735198..90af4e3022 100644 --- a/src/event.rs +++ b/src/event.rs @@ -34,6 +34,7 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_liquidity::lsps2::utils::compute_opening_fee; use lightning_types::payment::{PaymentHash, PaymentPreimage}; +use crate::channel::SpliceRetrier; use crate::config::{may_announce_channel, Config, PEER_RECONNECTION_INTERVAL}; use crate::connection::ConnectionManager; use crate::data_store::DataStoreUpdateResult; @@ -283,7 +284,11 @@ pub enum Event { /// The outpoint of the channel's splice funding transaction. new_funding_txo: OutPoint, }, - /// A channel splice negotiation round has failed. + /// A channel splice has failed and is no longer being pursued. + /// + /// A recoverable failure of a user-initiated splice (e.g. the peer disconnecting + /// mid-negotiation) is retried automatically, including across restarts; this event is emitted + /// only once the splice is given up on. SpliceNegotiationFailed { /// The `channel_id` of the channel. channel_id: ChannelId, @@ -541,6 +546,7 @@ where onion_messenger: Arc, om_mailbox: Option>, prober: Option>, + splice_retrier: Arc>, runtime: Arc, logger: L, config: Arc, @@ -559,7 +565,8 @@ where peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, prober: Option>, - runtime: Arc, logger: L, config: Arc, + splice_retrier: Arc>, runtime: Arc, logger: L, + config: Arc, ) -> Self { Self { event_queue, @@ -577,6 +584,7 @@ where onion_messenger, om_mailbox, prober, + splice_retrier, runtime, logger, config, @@ -1649,6 +1657,10 @@ where .handle_channel_ready(user_channel_id, &channel_id, &counterparty_node_id) .await; + self.splice_retrier + .on_channel_ready(UserChannelId(user_channel_id), funding_txo) + .await; + let event = Event::ChannelReady { channel_id, user_channel_id: UserChannelId(user_channel_id), @@ -1672,6 +1684,8 @@ where } => { log_info!(self.logger, "Channel {} closed due to: {}", channel_id, reason); + self.splice_retrier.on_channel_closed(UserChannelId(user_channel_id)).await; + // `counterparty_node_id` has been set on every `ChannelClosed` since LDK 0.0.117. let counterparty_node_id = counterparty_node_id .expect("counterparty_node_id is always set since LDK 0.0.117"); @@ -1983,15 +1997,27 @@ where channel_id, user_channel_id, counterparty_node_id, - .. + reason, + contribution, } => { log_info!( self.logger, - "Channel {} with counterparty {} splice negotiation failed", + "Channel {} with counterparty {} splice negotiation failed: {}", channel_id, counterparty_node_id, + reason, ); + // A user-initiated splice is retried automatically, including across restarts; + // surface the failure only once it is given up on. + let surface = self + .splice_retrier + .on_negotiation_failed(UserChannelId(user_channel_id), reason, contribution) + .await; + if !surface { + return Ok(()); + } + let event = Event::SpliceNegotiationFailed { channel_id, user_channel_id: UserChannelId(user_channel_id), diff --git a/src/lib.rs b/src/lib.rs index a14bcd42dd..1d1539dbba 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -661,6 +661,8 @@ impl Node { let splice_retrier = Arc::new(SpliceRetrier::new( Arc::clone(&self.channel_manager), + Arc::clone(&self.wallet), + Arc::clone(&self.fee_estimator), Arc::clone(&self.pending_payment_store), Arc::clone(&self.event_queue), Arc::clone(&self.logger), @@ -682,6 +684,7 @@ impl Node { Arc::clone(&self.onion_messenger), self.om_mailbox.clone(), self.prober.clone(), + Arc::clone(&splice_retrier), Arc::clone(&self.runtime), Arc::clone(&self.logger), Arc::clone(&self.config), From bc76ab8665a17604828a55729ddf7f3d5d1d80d1 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 09:51:44 -0500 Subject: [PATCH 10/11] Apply funding status updates atomically with the funding-type check apply_funding_status_update fetched the payment record, checked that it was a classified funding payment, merged in the new confirmation status, and wrote the result back as separate store operations. A classification racing in between -- merging tx_type and the contribution-derived figures into the record -- would be overwritten by the stale snapshot. Perform the check and the merge under the payment store's mutation lock so no write can interleave, and skip persisting when the merge changes nothing. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 72 ++++++++++++++++++++++++++++++----------------- 1 file changed, 46 insertions(+), 26 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index b64e227323..a35aa5b63d 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1650,34 +1650,54 @@ impl Wallet { async fn apply_funding_status_update( &self, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus, ) -> Result { - let Some(mut payment) = self.payment_store.get(&payment_id) else { + // The funding-type gate, the candidate lookup, and the write share the store's mutation + // lock: against a separate `get`, a classification merging in between would have its + // `tx_type` and contribution figures clobbered by this stale snapshot. + let mut handled = None; + self.payment_store + .mutate(&payment_id, |existing| { + let payment = existing?; + let tx_type = match &payment.kind { + PaymentKind::Onchain { + tx_type: + tx_type @ Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. }, + ), + .. + } => tx_type.clone(), + _ => 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) = self.pending_payment_store.get(&payment_id) { + if let Some(candidate) = pending.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 }; + + // Merge through the update machinery so its rules (e.g. which fields a merge may + // touch) keep applying, and skip the write when nothing changed. + let mut merged = payment.clone(); + if merged.update(target.to_update()) { + handled = Some(merged.clone()); + Some(merged) + } else { + handled = Some(payment.clone()); + None + } + }) + .await?; + let Some(payment) = handled else { return Ok(false); }; - let tx_type = match &payment.kind { - PaymentKind::Onchain { - tx_type: - tx_type @ Some( - TransactionType::Funding { .. } - | TransactionType::InteractiveFunding { .. }, - ), - .. - } => tx_type.clone(), - _ => return Ok(false), - }; - // 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.) - if let Some(pending) = self.pending_payment_store.get(&payment_id) { - if let Some(candidate) = pending.candidate(event_txid) { - payment.amount_msat = candidate.amount_msat; - payment.fee_paid_msat = candidate.fee_paid_msat; - } - } - - payment.kind = - PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type }; - self.payment_store.insert_or_update(payment.clone()).await?; // 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 // the same dual-write the default `TxConfirmed` path performs; an empty conflicting-txids From 7265f8cd81d0780369760fe7d37041df976407aa Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 2 Jul 2026 00:45:11 -0500 Subject: [PATCH 11/11] Test splice resumption across restarts and document automatic retry Add integration coverage for resuming a dropped splice: splice_resumed_after_restart initiates a splice-out while disconnected, restarts the node before anything is negotiated, and asserts the reconciler resumes and completes the splice -- and that a second restart does not resubmit the now-locked splice. splice_rbf_resumed_after_restart does the same for a fee bump. Also cover the id-agreement race: splice_payment_tracked_across_restart_before_lock stops the node right after splice negotiation, lets the counterparty's broadcast confirm while it is down, and asserts after restart that wallet sync and classification -- landing in either order -- produce exactly one payment record, keyed by the splice-time id rather than a txid-derived one, through to Succeeded. Document on splice_in, splice_out, and bump_channel_funding_fee that the splice is retried automatically across restarts until it completes or is given up on. Generated with assistance from Claude Code. Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Fable 5 --- src/lib.rs | 16 ++ tests/integration_tests_rust.rs | 303 +++++++++++++++++++++++++++++++- 2 files changed, 318 insertions(+), 1 deletion(-) diff --git a/src/lib.rs b/src/lib.rs index 1d1539dbba..a6a5dbdb25 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1904,6 +1904,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-in will be marked as an outbound payment, but @@ -1928,6 +1932,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-in will be marked as an outbound payment, but @@ -1944,6 +1952,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-out will be marked as an inbound payment if @@ -2041,6 +2053,10 @@ impl Node { /// Fee-bumps the pending splice on a channel by replacing its in-flight funding transaction /// (RBF). The splice's amount and destination are preserved; only the fee rate is raised. /// Errors if the channel has no pending splice to bump. + /// + /// The fee bump is retried automatically, including across restarts, until it either completes + /// or fails for a reason retrying cannot address, at which point + /// [`Event::SpliceNegotiationFailed`] is emitted. pub fn bump_channel_funding_fee( &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, ) -> Result<(), Error> { diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 62f9615b28..5863c54a72 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -2003,7 +2003,9 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { }, } assert_eq!(payment.status, PaymentStatus::Pending); - // Only one Onchain Pending payment for this splice attempt (not one per candidate). + // Only one Onchain Pending payment for this splice attempt (not one per candidate). This also + // guards the intent-clobber fix: had the sync above cleared this splice's live intent, the + // bump would not have found it and would have minted a second record under a fresh PaymentId. let splice_payments = node_b.list_payments_with_filter(|p| { p.direction == PaymentDirection::Outbound && matches!(p.kind, PaymentKind::Onchain { .. }) @@ -2221,6 +2223,305 @@ async fn splice_payment_reorged_to_unconfirmed() { node_b.stop().unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_resumed_after_restart() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let onchain_balance_before_sat = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Initiate a splice-out while disconnected: LDK accepts the contribution but cannot make + // progress before the restart below drops it, having neither negotiated nor persisted + // anything. Only the persisted splice intent allows resuming the splice. + node_a.disconnect(node_b.node_id()).unwrap(); + let address = node_a.onchain_payment().new_address().unwrap(); + node_a.splice_out(&user_channel_id_a, node_b.node_id(), &address, 500_000).unwrap(); + + let onchain_balance_before_sat = node_a.list_balances().total_onchain_balance_sats; + node_a.stop().unwrap(); + onchain_balance_before_sat + }; + + // On restart, the reconciler resubmits the splice, which proceeds once the peer connects. + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + + let txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + expect_splice_negotiated_event!(node_b, node_a.node_id()); + + wait_for_tx(&electrsd.client, txo.txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + assert!( + node_a.list_balances().total_onchain_balance_sats > onchain_balance_before_sat + 400_000, + "resumed splice-out should have moved ~500k sats to the on-chain balance", + ); + + // The locked splice cleared the intent, so another restart must not resubmit it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "completed splice should not be resubmitted"); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_rbf_resumed_after_restart() { + // Use a custom bitcoind config with a lower incrementalrelayfee so that the +25 sat/kwu + // (0.1 sat/vB) RBF feerate bump satisfies BIP125's absolute fee increase requirement. + let bitcoind_exe = std::env::var("BITCOIND_EXE") + .ok() + .or_else(|| corepc_node::downloaded_exe_path().ok()) + .expect( + "you need to provide an env var BITCOIND_EXE or specify a bitcoind version feature", + ); + let mut bitcoind_conf = corepc_node::Conf::default(); + bitcoind_conf.network = "regtest"; + bitcoind_conf.args.push("-rest"); + bitcoind_conf.args.push("-incrementalrelayfee=0.00000100"); + let bitcoind = BitcoinD::with_conf(bitcoind_exe, &bitcoind_conf).unwrap(); + + let electrs_exe = std::env::var("ELECTRS_EXE") + .ok() + .or_else(electrsd::downloaded_exe_path) + .expect("you need to provide env var ELECTRS_EXE or specify an electrsd version feature"); + let mut electrsd_conf = electrsd::Conf::default(); + electrsd_conf.http_enabled = true; + electrsd_conf.network = "regtest"; + let electrsd = ElectrsD::with_conf(electrs_exe, &bitcoind, &electrsd_conf).unwrap(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let original_txo = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Negotiate a splice but leave its transaction unconfirmed so it can be fee-bumped. + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 500_000).unwrap(); + let original_txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + expect_splice_negotiated_event!(node_b, node_a.node_id()); + wait_for_tx(&electrsd.client, original_txo.txid).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + // Bump the fee while disconnected and restart before anything could be negotiated: only + // the persisted intent knows about the fee bump, while LDK still has the negotiated + // splice at the original feerate. + node_a.disconnect(node_b.node_id()).unwrap(); + node_a.bump_channel_funding_fee(&user_channel_id_a, node_b.node_id()).unwrap(); + node_a.stop().unwrap(); + original_txo + }; + + // On restart, the reconciler sees that the negotiated splice is still at a lower feerate + // than the persisted fee-bump intent and resubmits the bump. + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + + let rbf_txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + expect_splice_negotiated_event!(node_b, node_a.node_id()); + assert_ne!(original_txo, rbf_txo, "resubmitted RBF should produce a different funding txo"); + + // Restarting again must not resubmit the bump: the negotiated splice now carries it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "carried-out fee bump should not be resubmitted"); + + wait_for_tx(&electrsd.client, rbf_txo.txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // The locked fee bump cleared its intent, so a further restart must not resubmit it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "locked fee bump should not be resubmitted"); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_payment_tracked_across_restart_before_lock() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let splice_txid = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 500_000).unwrap(); + let txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + expect_splice_negotiated_event!(node_b, node_a.node_id()); + + // Stop node_a as soon as the splice is negotiated. node_b broadcasts the transaction + // either way, so it reaches the chain while node_a is offline. Depending on timing, + // node_a may or may not have classified its own broadcast into a payment record before + // stopping; the assertions below must hold in both cases. + node_a.stop().unwrap(); + txo.txid + }; + + // Confirm the splice while node_a is offline, but keep it short of the depth at which it + // locks, so node_a restarts with its splice intent still live. + wait_for_tx(&electrsd.client, splice_txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + + // After the restart, wallet sync and classification must agree on the splice-time + // `PaymentId` no matter which of them sees the confirmed transaction first: exactly one + // payment record, and not one keyed by a txid-derived id. + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + + let splice_payments = |node: &Node| { + node.list_payments_with_filter( + |p| matches!(p.kind, PaymentKind::Onchain { txid, .. } if txid == splice_txid), + ) + }; + let payments = splice_payments(&node_a); + assert_eq!( + payments.len(), + 1, + "expected exactly one payment record for the splice, got {}: {:#?}", + payments.len(), + payments, + ); + assert_ne!( + payments[0].id, + PaymentId(splice_txid.to_byte_array()), + "the splice payment must keep its splice-time id, not a txid-derived fallback", + ); + assert_eq!(payments[0].status, PaymentStatus::Pending); + + // Reconnect and let the splice lock: the single record graduates instead of gaining a + // duplicate. + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 5).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + let payments = splice_payments(&node_a); + assert_eq!( + payments.len(), + 1, + "expected exactly one payment record after the splice locked, got {}: {:#?}", + payments.len(), + payments, + ); + assert_eq!(payments[0].status, PaymentStatus::Succeeded); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn splice_in_rbf_joins_counterparty_splice() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();