From 98b11a5856593b778de73d585b6d65c418291365 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 1 Jul 2026 18:49:09 -0500 Subject: [PATCH 1/5] 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 2/5] 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 3/5] 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 4/5] 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 5/5] 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(()) }