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