From 98b11a5856593b778de73d585b6d65c418291365 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 1 Jul 2026 18:49:09 -0500 Subject: [PATCH 01/13] Preserve funding-payment confirmation state on late reclassification Funding broadcasts are classified into payment records off the broadcaster's queue, which can run after wallet sync has already recorded the transaction -- for instance when the counterparty's broadcast of the funding transaction is observed by wallet sync first. In that case the classification overwrote a record wallet sync had already advanced, downgrading a confirmed or graduated funding payment back to unconfirmed/pending. Merge only the classification and our contribution figures into an existing record, leaving the confirmation state that the wallet-sync events own in place. Raised by Codex in the review of #888. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/payment/store.rs | 115 +++++++++++++++++++++++++++++++++++++++++++ src/wallet/mod.rs | 28 +++++++++-- 2 files changed, 139 insertions(+), 4 deletions(-) diff --git a/src/payment/store.rs b/src/payment/store.rs index d2b92747a2..df2e9dc099 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -712,6 +712,33 @@ impl PaymentDetailsUpdate { tx_type: None, } } + + /// Builds an update that merges a freshly-classified funding payment's classification + /// (`tx_type`), broadcast txid, and our contribution figures (amount/fee) into an existing + /// record, while leaving the top-level [`PaymentStatus`] and the on-chain + /// [`ConfirmationStatus`] untouched. + /// + /// Funding classification runs off the broadcaster queue and can land *after* wallet sync has + /// already advanced a record's confirmation state (e.g. when LDK re-broadcasts a still-pending + /// funding transaction on restart, or when the counterparty's broadcast is observed first). + /// Merging only the funding-specific fields keeps such a late classification from downgrading a + /// `Confirmed`/`Succeeded` payment back to `Unconfirmed`/`Pending`; the confirmation state is + /// owned by the wallet-sync events instead. + /// + /// The txid and figures are taken from the freshly broadcast (active) candidate. LDK only + /// re-broadcasts the active/confirmed funding candidate, so for an already-confirmed record + /// these equal what graduation stamped and the overwrite is a no-op; we rely on that invariant + /// rather than gating the txid/amount/fee merge on the stored confirmation state. + pub(crate) fn funding_reclassification(details: PaymentDetails) -> Self { + let mut update = Self::new(details.id); + update.amount_msat = Some(details.amount_msat); + update.fee_paid_msat = Some(details.fee_paid_msat); + if let PaymentKind::Onchain { txid, tx_type, .. } = details.kind { + update.txid = Some(txid); + update.tx_type = Some(tx_type); + } + update + } } impl From<&PaymentDetails> for PaymentDetailsUpdate { @@ -1022,6 +1049,94 @@ mod tests { } } + #[test] + fn funding_reclassification_does_not_downgrade_an_advanced_record() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // A splice funding payment wallet sync has already advanced to Succeeded/Confirmed. + let txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(txid.to_byte_array()); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let advanced = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: tx_type.clone(), + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Succeeded, + ); + + // A fresh funding classification for the same payment is always Pending/Unconfirmed. + let fresh = PaymentDetails::new( + id, + PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The naive full update `insert_or_update` applied before the fix downgrades both the + // top-level status and the on-chain confirmation status — the bug Codex flagged. + let mut downgraded = advanced.clone(); + downgraded.update((&fresh).into()); + assert_eq!( + downgraded.status, + PaymentStatus::Pending, + "a full update from a fresh classification downgrades the top-level status", + ); + assert!( + matches!( + downgraded.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + ), + "a full update from a fresh classification downgrades the confirmation status", + ); + + // The narrowed reclassification update merges only the funding fields and preserves the + // advanced confirmation state that wallet sync owns. + let mut merged = advanced.clone(); + merged.update(PaymentDetailsUpdate::funding_reclassification(fresh)); + assert_eq!( + merged.status, + PaymentStatus::Succeeded, + "reclassification must not downgrade the top-level status", + ); + assert!( + matches!( + merged.kind, + PaymentKind::Onchain { + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } + ), + "reclassification must preserve the confirmation status and keep the funding tx_type", + ); + // The contribution-derived figures from the fresh classification ARE merged in, replacing + // the existing record's: they are authoritative (the wallet can't recompute our share of a + // shared funding output), so the merge must carry them. + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); + } + #[derive(Clone, Debug, PartialEq, Eq)] struct LegacyBolt11JitKind { hash: PaymentHash, diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f8d9d521eb..5b53ef292f 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -56,7 +56,8 @@ use persist::KVStoreWalletPersister; use crate::config::Config; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; -use crate::payment::store::ConfirmationStatus; +use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; +use crate::payment::store::{ConfirmationStatus, PaymentDetailsUpdate}; use crate::payment::{ FundingTxCandidate, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, PendingPaymentDetails, TransactionType, @@ -1398,9 +1399,28 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { - self.payment_store.insert_or_update(details.clone()).await?; - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.insert_or_update(pending).await?; + if !self.payment_store.contains_key(&details.id) { + // First time we record this funding payment: store it and index it for graduation. + self.payment_store.insert_or_update(details.clone()).await?; + let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); + self.pending_payment_store.insert_or_update(pending).await?; + } else { + // An earlier candidate or a racing wallet sync already recorded this payment. Merge only + // the classification (`tx_type`) and our contribution figures, which the wallet can't + // recompute; the confirmation state is owned by wallet-sync events, so a late + // classification must not move it (which would downgrade an already-Confirmed/Succeeded + // record). `update` is a no-op when the entry is absent, so the pending index is not + // re-created for a payment the graduation path already removed. + let update = PaymentDetailsUpdate::funding_reclassification(details); + let pending_update = PendingPaymentDetailsUpdate { + id: update.id, + payment_update: Some(update.clone()), + conflicting_txids: None, + candidates, + }; + self.payment_store.update(update).await?; + self.pending_payment_store.update(pending_update).await?; + } Ok(()) } From cb99564c8065a92cf44ffb9f208650ba1c8b125f Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 28 Jul 2026 18:24:00 -0500 Subject: [PATCH 02/13] f - Make funding-payment classification write atomic with its existence check Co-Authored-By: Claude --- src/data_store.rs | 149 +++++++++++++++++++++++++++ src/payment/pending_payment_store.rs | 74 +++++++++++++ src/payment/store.rs | 105 +++++++++++++++++++ src/wallet/mod.rs | 55 ++++++---- 4 files changed, 362 insertions(+), 21 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index b1ed816df9..b7748b172e 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -40,6 +40,13 @@ pub(crate) enum DataStoreUpdateResult { NotFound, } +#[derive(PartialEq, Eq, Debug, Clone, Copy)] +pub(crate) enum DataStoreUpdateOrInsertResult { + Inserted, + Updated, + Unchanged, +} + pub(crate) struct DataStore where L::Target: LdkLogger, @@ -81,6 +88,11 @@ where Ok(updated) } + /// Like [`Self::insert`], but when an entry with the object's id already exists, merges the + /// object's full update ([`StorableObject::to_update`]) into it instead of replacing it. + /// + /// Unlike [`Self::update_or_insert`], the caller does not choose what is merged into an + /// existing entry: the full update is always applied. pub(crate) async fn insert_or_update(&self, object: SO) -> Result { let _guard = self.mutation_lock.lock().await; @@ -170,6 +182,47 @@ where Ok(DataStoreUpdateResult::Updated) } + /// Applies `update` when an object with its id already exists, or inserts `object` when none + /// does. + /// + /// Like [`Self::update`], but falls back to inserting `object` instead of returning + /// [`DataStoreUpdateResult::NotFound`]. Unlike [`Self::insert_or_update`], the caller chooses + /// exactly what is merged into an existing entry: `update` may carry less than the full + /// object. + /// + /// The existence check and the write share one critical section of the mutation lock, so a + /// concurrent writer cannot land in between and later have its state clobbered by the insert + /// fallback — the check-then-act race that separate [`Self::contains_key`] + + /// [`Self::insert_or_update`] calls reintroduce. + pub(crate) async fn update_or_insert( + &self, update: SO::Update, object: SO, + ) -> Result { + debug_assert!(update.id() == object.id(), "update and object must share an id"); + let _guard = self.mutation_lock.lock().await; + + let id = update.id(); + let (data_to_persist, result) = { + let locked_objects = self.objects.lock().expect("lock"); + match locked_objects.get(&id) { + Some(existing_object) => { + let mut updated_object = existing_object.clone(); + if updated_object.update(update) { + (Some(updated_object), DataStoreUpdateOrInsertResult::Updated) + } else { + (None, DataStoreUpdateOrInsertResult::Unchanged) + } + }, + None => (Some(object), DataStoreUpdateOrInsertResult::Inserted), + } + }; + + if let Some(object) = data_to_persist { + self.persist(&object).await?; + self.objects.lock().expect("lock").insert(id, object); + } + Ok(result) + } + /// Returns in-memory objects matching `f`. /// /// The async mutation lock serializes writers, but this synchronous reader cannot wait on it. @@ -403,6 +456,102 @@ mod tests { assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await); } + #[tokio::test] + async fn update_or_insert_inserts_when_absent() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let primary_namespace = "datastore_test_primary".to_string(); + let secondary_namespace = "datastore_test_secondary".to_string(); + let data_store: DataStore> = DataStore::new( + Vec::new(), + primary_namespace.clone(), + secondary_namespace.clone(), + Arc::clone(&store), + logger, + ); + + let id = TestObjectId { id: [42u8; 4] }; + let object = TestObject { id, data: [23u8; 3] }; + let update = TestObjectUpdate { id, data: [25u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Inserted), + data_store.update_or_insert(update, object).await + ); + + // The insert path stores the fallback object as-is; the update is not applied to it. + assert_eq!(Some(object), data_store.get(&id)); + let store_key = id.encode_to_hex_str(); + assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key) + .await + .is_ok()); + } + + #[tokio::test] + async fn update_or_insert_applies_update_when_present() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store: DataStore> = DataStore::new( + vec![existing_object], + "datastore_test_primary".to_string(), + "datastore_test_secondary".to_string(), + store, + logger, + ); + + // When an entry exists, only the update is applied; the fallback object must not replace + // it. + let update = TestObjectUpdate { id, data: [24u8; 3] }; + let object = TestObject { id, data: [99u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Updated), + data_store.update_or_insert(update, object).await + ); + assert_eq!(data_store.get(&id).unwrap().data, [24u8; 3]); + } + + #[tokio::test] + async fn update_or_insert_returns_unchanged_without_persisting() { + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + // A no-op update returns `Unchanged` without attempting to persist (the store fails all + // writes) and without falling back to the object. + let update = TestObjectUpdate { id, data: [23u8; 3] }; + let object = TestObject { id, data: [99u8; 3] }; + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Unchanged), + data_store.update_or_insert(update, object).await + ); + assert_eq!(Some(existing_object), data_store.get(&id)); + } + + #[tokio::test] + async fn update_or_insert_does_not_mutate_memory_if_persist_fails() { + let existing_id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id: existing_id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + let update = TestObjectUpdate { id: existing_id, data: [24u8; 3] }; + let object = TestObject { id: existing_id, data: [24u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.update_or_insert(update, object).await + ); + assert_eq!(Some(existing_object), data_store.get(&existing_id)); + + let new_id = TestObjectId { id: [55u8; 4] }; + let new_object = TestObject { id: new_id, data: [34u8; 3] }; + let new_update = TestObjectUpdate { id: new_id, data: [34u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.update_or_insert(new_update, new_object).await + ); + assert!(data_store.get(&new_id).is_none()); + } + #[tokio::test] async fn insert_or_update_does_not_mutate_memory_if_persist_fails() { let existing_id = TestObjectId { id: [42u8; 4] }; diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index f5f2fa40a2..b822427dfb 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -242,4 +242,78 @@ mod tests { "current txid must not remain in its own conflict list" ); } + + #[test] + fn funding_classification_pending_update_preserves_mirrored_confirmation() { + use bitcoin::BlockHash; + + use crate::payment::store::PaymentDetailsUpdate; + + let txid = test_txid(7); + let payment_id = PaymentId(txid.to_byte_array()); + + // A pending entry wallet sync has already mirrored a confirmation into (via + // `apply_funding_status_update`) while classification was still writing its two stores. + let confirmed_details = PaymentDetails::new( + payment_id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + let mirrored = PendingPaymentDetails::new(confirmed_details, Vec::new(), Vec::new()); + + // A fresh classification is always Unconfirmed and carries the candidate history; its + // figures are the active candidate's. + let fresh = pending_onchain_payment(payment_id, txid); + let candidates = vec![FundingTxCandidate { + txid, + amount_msat: fresh.amount_msat, + fee_paid_msat: fresh.fee_paid_msat, + }]; + + // The old fresh-insert path merged the full fresh record, downgrading the mirrored + // confirmation. + let mut downgraded = mirrored.clone(); + let full_update = + PendingPaymentDetails::new(fresh.clone(), Vec::new(), candidates.clone()).to_update(); + assert!(downgraded.update(full_update)); + assert!( + matches!( + downgraded.details.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + ), + "a full merge of a fresh classification downgrades a mirrored confirmation", + ); + + // The narrow classification update merges the figures and candidates while preserving the + // confirmation state wallet sync owns. + let mut merged = mirrored.clone(); + let narrow_update = PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)), + conflicting_txids: None, + candidates: candidates.clone(), + }; + assert!(merged.update(narrow_update)); + assert!( + matches!( + merged.details.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + ), + "a narrow classification update must not downgrade a mirrored confirmation", + ); + assert_eq!(merged.candidates, candidates); + assert_eq!(merged.details.amount_msat, Some(1_000)); + assert_eq!(merged.details.fee_paid_msat, Some(100)); + } } diff --git a/src/payment/store.rs b/src/payment/store.rs index df2e9dc099..17819b03a6 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -1137,6 +1137,111 @@ mod tests { assert_eq!(merged.fee_paid_msat, Some(500)); } + #[tokio::test] + async fn funding_classification_update_or_insert_preserves_advanced_record() { + use bitcoin::hashes::Hash; + use lightning::util::test_utils::TestLogger; + use std::str::FromStr; + use std::sync::Arc; + + use crate::data_store::{DataStore, DataStoreUpdateOrInsertResult}; + use crate::io::test_utils::InMemoryStore; + use crate::types::{DynStore, DynStoreWrapper}; + + let txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(txid.to_byte_array()); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + // A funding payment wallet sync has already advanced to Succeeded/Confirmed. + let advanced = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: tx_type.clone(), + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Succeeded, + ); + // A fresh funding classification for the same payment is always Pending/Unconfirmed. + let fresh = PaymentDetails::new( + id, + PaymentKind::Onchain { txid, status: ConfirmationStatus::Unconfirmed, tx_type }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + let new_store = |seed: Vec| { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + DataStore::>::new( + seed, + "payment_test_primary".to_string(), + "payment_test_secondary".to_string(), + store, + logger, + ) + }; + + // The pre-fix fresh-insert path — a full `insert_or_update` merge landing after a racing + // wallet sync already advanced the record — downgrades it. + let store = new_store(vec![advanced.clone()]); + store.insert_or_update(fresh.clone()).await.unwrap(); + let downgraded = store.get(&id).unwrap(); + assert_eq!( + downgraded.status, + PaymentStatus::Pending, + "a full merge of a fresh classification downgrades an advanced record", + ); + + // `update_or_insert` applies only the narrow reclassification when a record exists — no + // matter when it appeared — preserving the confirmation state wallet sync owns while + // still merging the contribution-derived figures. + let store = new_store(vec![advanced.clone()]); + let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Updated), + store.update_or_insert(update, fresh.clone()).await + ); + let merged = store.get(&id).unwrap(); + assert_eq!(merged.status, PaymentStatus::Succeeded); + assert!(matches!( + merged.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + )); + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); + + // And it inserts the fresh details when no record exists yet. + let store = new_store(Vec::new()); + let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); + assert_eq!( + Ok(DataStoreUpdateOrInsertResult::Inserted), + store.update_or_insert(update, fresh).await + ); + let inserted = store.get(&id).unwrap(); + assert_eq!(inserted.status, PaymentStatus::Pending); + assert!(matches!( + inserted.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + )); + } + #[derive(Clone, Debug, PartialEq, Eq)] struct LegacyBolt11JitKind { hash: PaymentHash, diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 5b53ef292f..2e6c41a436 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; +use crate::data_store::DataStoreUpdateOrInsertResult; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -1399,27 +1400,39 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { - if !self.payment_store.contains_key(&details.id) { - // First time we record this funding payment: store it and index it for graduation. - self.payment_store.insert_or_update(details.clone()).await?; - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.insert_or_update(pending).await?; - } else { - // An earlier candidate or a racing wallet sync already recorded this payment. Merge only - // the classification (`tx_type`) and our contribution figures, which the wallet can't - // recompute; the confirmation state is owned by wallet-sync events, so a late - // classification must not move it (which would downgrade an already-Confirmed/Succeeded - // record). `update` is a no-op when the entry is absent, so the pending index is not - // re-created for a payment the graduation path already removed. - let update = PaymentDetailsUpdate::funding_reclassification(details); - let pending_update = PendingPaymentDetailsUpdate { - id: update.id, - payment_update: Some(update.clone()), - conflicting_txids: None, - candidates, - }; - self.payment_store.update(update).await?; - self.pending_payment_store.update(pending_update).await?; + // Deciding between a fresh insert and a merge must be atomic with the write: a racing + // wallet sync can record and advance this payment between a separate existence check and + // the write, and a full merge of the fresh Pending/Unconfirmed details would then + // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds + // the store's mutation lock across the whole decision: when a record exists — no matter + // when it appeared — only the classification (`tx_type`), the broadcast txid, and our + // contribution figures are merged, which the wallet can't recompute; otherwise the fresh + // details are inserted. + let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); + let pending_update = PendingPaymentDetailsUpdate { + id: update.id, + payment_update: Some(update.clone()), + conflicting_txids: None, + candidates: candidates.clone(), + }; + match self.payment_store.update_or_insert(update, details.clone()).await? { + DataStoreUpdateOrInsertResult::Inserted => { + // First time we record this funding payment: index it for graduation. Wallet sync + // can still land between the payment-store write above and this one and mirror an + // advanced confirmation into the pending store, so this write makes the same + // atomic decision: merge narrowly into an entry that appeared, insert the fresh + // one otherwise. + let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); + self.pending_payment_store.update_or_insert(pending_update, pending).await?; + }, + DataStoreUpdateOrInsertResult::Updated | DataStoreUpdateOrInsertResult::Unchanged => { + // An earlier candidate or a racing wallet sync already recorded this payment. + // `update` is a no-op when the pending entry is absent, so the index is not + // re-created for a payment the graduation path already removed. (A graduated + // payment always has a payment-store record, so it cannot take the `Inserted` + // branch above and be re-indexed.) + self.pending_payment_store.update(pending_update).await?; + }, } Ok(()) } From 0e44736e3a0cc8be35a73b83d2ca6ff8098920f7 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 28 Jul 2026 18:45:05 -0500 Subject: [PATCH 03/13] f - Keep a confirmed funding record's txid and figures on late classification Co-Authored-By: Claude --- src/payment/pending_payment_store.rs | 8 +- src/payment/store.rs | 155 ++++++++++++++++++++++----- src/wallet/mod.rs | 6 +- 3 files changed, 135 insertions(+), 34 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index b822427dfb..2257de43cb 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -295,8 +295,8 @@ mod tests { "a full merge of a fresh classification downgrades a mirrored confirmation", ); - // The narrow classification update merges the figures and candidates while preserving the - // confirmation state wallet sync owns. + // The narrow classification update merges the candidates while preserving the + // confirmation state wallet sync owns and the confirmed candidate's figures. let mut merged = mirrored.clone(); let narrow_update = PendingPaymentDetailsUpdate { id: payment_id, @@ -313,7 +313,7 @@ mod tests { "a narrow classification update must not downgrade a mirrored confirmation", ); assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(1_000)); - assert_eq!(merged.details.fee_paid_msat, Some(100)); + assert_eq!(merged.details.amount_msat, Some(2_000_000)); + assert_eq!(merged.details.fee_paid_msat, Some(999)); } } diff --git a/src/payment/store.rs b/src/payment/store.rs index 17819b03a6..00fcffcff8 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -243,12 +243,24 @@ impl StorableObject for PaymentDetails { } } - if let Some(amount_opt) = update.amount_msat { - update_if_necessary!(self.amount_msat, amount_opt); - } + // Once an on-chain record is confirmed, its txid and figures describe the candidate that + // confirmed, which need not be the last one broadcast. An update that doesn't assert the + // confirmation state was built without knowing it — e.g. a late funding classification + // whose candidate lost to the counterparty's broadcast — so it must not move them. + let keep_confirmed_figures = update.confirmation_status.is_none() + && matches!( + self.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + ); + + if !keep_confirmed_figures { + if let Some(amount_opt) = update.amount_msat { + update_if_necessary!(self.amount_msat, amount_opt); + } - if let Some(fee_paid_msat_opt) = update.fee_paid_msat { - update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt); + if let Some(fee_paid_msat_opt) = update.fee_paid_msat { + update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt); + } } if let Some(skimmed_fee_msat) = update.counterparty_skimmed_fee_msat { @@ -278,7 +290,7 @@ impl StorableObject for PaymentDetails { if let Some(tx_id) = update.txid { match self.kind { - PaymentKind::Onchain { ref mut txid, .. } => { + PaymentKind::Onchain { ref mut txid, .. } if !keep_confirmed_figures => { update_if_necessary!(*txid, tx_id); }, _ => {}, @@ -719,16 +731,16 @@ impl PaymentDetailsUpdate { /// [`ConfirmationStatus`] untouched. /// /// Funding classification runs off the broadcaster queue and can land *after* wallet sync has - /// already advanced a record's confirmation state (e.g. when LDK re-broadcasts a still-pending - /// funding transaction on restart, or when the counterparty's broadcast is observed first). - /// Merging only the funding-specific fields keeps such a late classification from downgrading a - /// `Confirmed`/`Succeeded` payment back to `Unconfirmed`/`Pending`; the confirmation state is - /// owned by the wallet-sync events instead. + /// already advanced a record's confirmation state (e.g. when the counterparty's broadcast of + /// the funding transaction is observed first). Merging only the funding-specific fields keeps + /// such a late classification from downgrading a `Confirmed`/`Succeeded` payment back to + /// `Unconfirmed`/`Pending`; the confirmation state is owned by the wallet-sync events instead. /// - /// The txid and figures are taken from the freshly broadcast (active) candidate. LDK only - /// re-broadcasts the active/confirmed funding candidate, so for an already-confirmed record - /// these equal what graduation stamped and the overwrite is a no-op; we rely on that invariant - /// rather than gating the txid/amount/fee merge on the stored confirmation state. + /// The txid and figures are taken from the freshly broadcast (active) candidate, so they only + /// apply while the record is unconfirmed. Once a candidate confirms, the record's txid and + /// figures describe that candidate — which need not be the one being classified (e.g. the + /// counterparty broadcast an earlier candidate and it won) — and [`PaymentDetails::update`] + /// leaves them in place for updates like this one that don't carry a confirmation state. pub(crate) fn funding_reclassification(details: PaymentDetails) -> Self { let mut update = Self::new(details.id); update.amount_msat = Some(details.amount_msat); @@ -1130,11 +1142,95 @@ mod tests { ), "reclassification must preserve the confirmation status and keep the funding tx_type", ); - // The contribution-derived figures from the fresh classification ARE merged in, replacing - // the existing record's: they are authoritative (the wallet can't recompute our share of a - // shared funding output), so the merge must carry them. - assert_eq!(merged.amount_msat, Some(1_000_000)); - assert_eq!(merged.fee_paid_msat, Some(500)); + // The confirmed record's figures describe the candidate that confirmed, so the late + // classification must not replace them either. + assert_eq!(merged.amount_msat, Some(2_000_000)); + assert_eq!(merged.fee_paid_msat, Some(999)); + } + + #[test] + fn funding_reclassification_keeps_confirmed_candidate_figures() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // A funding payment whose first candidate wallet sync has already seen confirm — e.g. the + // counterparty's broadcast of it was picked up before our own later candidate was + // classified. The record is unclassified (created by the sync fallthrough). + let confirmed_txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(confirmed_txid.to_byte_array()); + let confirmed = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // Our own, different (e.g. fee-bumped) candidate is classified late. + let late_txid = Txid::from_byte_array([9u8; 32]); + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let late = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: late_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type, + }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The confirmed record's txid and figures describe the candidate that confirmed; the late + // classification must not replace them with an unconfirmed candidate's. The + // classification itself (`tx_type`) still lands. + let mut classified = confirmed.clone(); + classified.update(PaymentDetailsUpdate::funding_reclassification(late.clone())); + assert!( + matches!( + classified.kind, + PaymentKind::Onchain { + txid, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } if txid == confirmed_txid + ), + "a late classification must set the tx_type but not replace a confirmed record's txid", + ); + assert_eq!(classified.amount_msat, Some(2_000_000)); + assert_eq!(classified.fee_paid_msat, Some(999)); + + // While the record is still unconfirmed, the freshly broadcast candidate is the active + // one, so its txid and figures do replace the stored ones (RBF rotation). + let mut unconfirmed = confirmed.clone(); + if let PaymentKind::Onchain { ref mut status, .. } = unconfirmed.kind { + *status = ConfirmationStatus::Unconfirmed; + } + unconfirmed.update(PaymentDetailsUpdate::funding_reclassification(late)); + assert!( + matches!(unconfirmed.kind, PaymentKind::Onchain { txid, .. } if txid == late_txid), + "classifying a new candidate of an unconfirmed record rotates the txid", + ); + assert_eq!(unconfirmed.amount_msat, Some(1_000_000)); + assert_eq!(unconfirmed.fee_paid_msat, Some(500)); } #[tokio::test] @@ -1159,7 +1255,8 @@ mod tests { channel_id: ChannelId([3u8; 32]), }], }); - // A funding payment wallet sync has already advanced to Succeeded/Confirmed. + // A funding payment wallet sync has already recorded (unclassified, via the default + // on-chain path) and advanced to Succeeded/Confirmed. let advanced = PaymentDetails::new( id, PaymentKind::Onchain { @@ -1169,7 +1266,7 @@ mod tests { height: 100, timestamp: 1, }, - tx_type: tx_type.clone(), + tx_type: None, }, Some(2_000_000), Some(999), @@ -1210,8 +1307,8 @@ mod tests { ); // `update_or_insert` applies only the narrow reclassification when a record exists — no - // matter when it appeared — preserving the confirmation state wallet sync owns while - // still merging the contribution-derived figures. + // matter when it appeared — setting the `tx_type` while preserving the confirmation + // state wallet sync owns and the confirmed candidate's figures. let store = new_store(vec![advanced.clone()]); let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); assert_eq!( @@ -1222,10 +1319,14 @@ mod tests { assert_eq!(merged.status, PaymentStatus::Succeeded); assert!(matches!( merged.kind, - PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + PaymentKind::Onchain { + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } )); - assert_eq!(merged.amount_msat, Some(1_000_000)); - assert_eq!(merged.fee_paid_msat, Some(500)); + assert_eq!(merged.amount_msat, Some(2_000_000)); + assert_eq!(merged.fee_paid_msat, Some(999)); // And it inserts the fresh details when no record exists yet. let store = new_store(Vec::new()); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 2e6c41a436..9e5039bfdb 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1405,9 +1405,9 @@ impl Wallet { // the write, and a full merge of the fresh Pending/Unconfirmed details would then // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds // the store's mutation lock across the whole decision: when a record exists — no matter - // when it appeared — only the classification (`tx_type`), the broadcast txid, and our - // contribution figures are merged, which the wallet can't recompute; otherwise the fresh - // details are inserted. + // when it appeared — only the classification (`tx_type`) and, while the record is still + // unconfirmed, the broadcast txid and our contribution figures are merged; otherwise the + // fresh details are inserted. let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); let pending_update = PendingPaymentDetailsUpdate { id: update.id, From 435caa6b8e58e289f905eeee9c6aa1f73fb87125 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 10:50:44 -0500 Subject: [PATCH 04/13] f - Merge figures when the classified candidate is the confirmed one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The keep-confirmed-figures guard overshot: when wallet sync recorded the confirmation first, the record carries the wallet's own view of amount/fee, which cannot represent our contribution to a shared funding output, and the guard then discarded the late classification's correct contribution-derived figures along with the losing-candidate updates it was meant to block. Let an update that names the confirmed txid move the figures — it describes the very candidate that confirmed — and have classification build its update from the candidate matching an already-confirmed record, mirroring what apply_funding_status_update reports when confirmation arrives after classification. The txid comparison happens under the store's mutation lock at apply time, so the unlocked snapshot read cannot misapply figures. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/payment/pending_payment_store.rs | 7 +- src/payment/store.rs | 92 +++++++++++++++-- src/wallet/mod.rs | 145 ++++++++++++++++++++++++++- 3 files changed, 228 insertions(+), 16 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 2257de43cb..a5994c8808 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -296,7 +296,8 @@ mod tests { ); // The narrow classification update merges the candidates while preserving the - // confirmation state wallet sync owns and the confirmed candidate's figures. + // confirmation state wallet sync owns. It names the confirmed txid, so its + // contribution-derived figures replace the mirrored wallet-view ones. let mut merged = mirrored.clone(); let narrow_update = PendingPaymentDetailsUpdate { id: payment_id, @@ -313,7 +314,7 @@ mod tests { "a narrow classification update must not downgrade a mirrored confirmation", ); assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(2_000_000)); - assert_eq!(merged.details.fee_paid_msat, Some(999)); + assert_eq!(merged.details.amount_msat, Some(1_000)); + assert_eq!(merged.details.fee_paid_msat, Some(100)); } } diff --git a/src/payment/store.rs b/src/payment/store.rs index 00fcffcff8..8eb787286c 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -246,11 +246,15 @@ impl StorableObject for PaymentDetails { // Once an on-chain record is confirmed, its txid and figures describe the candidate that // confirmed, which need not be the last one broadcast. An update that doesn't assert the // confirmation state was built without knowing it — e.g. a late funding classification - // whose candidate lost to the counterparty's broadcast — so it must not move them. + // whose candidate lost to the counterparty's broadcast — so it must not move them. The + // exception is an update naming the confirmed txid itself: its figures describe the very + // candidate that confirmed and correct the wallet-view amount/fee a sync-created record + // carries, which cannot represent our contribution to a shared funding output. let keep_confirmed_figures = update.confirmation_status.is_none() && matches!( self.kind, - PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } + PaymentKind::Onchain { txid, status: ConfirmationStatus::Confirmed { .. }, .. } + if update.txid != Some(txid) ); if !keep_confirmed_figures { @@ -1142,10 +1146,11 @@ mod tests { ), "reclassification must preserve the confirmation status and keep the funding tx_type", ); - // The confirmed record's figures describe the candidate that confirmed, so the late - // classification must not replace them either. - assert_eq!(merged.amount_msat, Some(2_000_000)); - assert_eq!(merged.fee_paid_msat, Some(999)); + // The late classification names the confirmed txid, so its contribution-derived figures + // replace the record's; only an update for a different candidate leaves them in place + // (covered by `funding_reclassification_keeps_confirmed_candidate_figures`). + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); } #[test] @@ -1233,6 +1238,74 @@ mod tests { assert_eq!(unconfirmed.fee_paid_msat, Some(500)); } + #[test] + fn funding_reclassification_merges_figures_for_the_confirmed_candidate() { + use bitcoin::hashes::Hash; + use std::str::FromStr; + + // Wallet sync confirmed the transaction before classification ran, so the record carries + // the wallet's own view of amount/fee, which cannot represent our contribution to a shared + // funding output. + let confirmed_txid = Txid::from_byte_array([7u8; 32]); + let id = PaymentId(confirmed_txid.to_byte_array()); + let mut record = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { + block_hash: BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + }, + tx_type: None, + }, + Some(2_000_000), + Some(999), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + // The late classification names the candidate that confirmed, so its contribution-derived + // figures are authoritative and must replace the wallet-view ones; only an update for a + // different (losing) candidate leaves a confirmed record's figures in place. + let tx_type = Some(TransactionType::InteractiveFunding { + channels: vec![Channel { + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([3u8; 32]), + }], + }); + let classified = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Unconfirmed, + tx_type, + }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ); + + assert!(record.update(PaymentDetailsUpdate::funding_reclassification(classified))); + assert!( + matches!( + record.kind, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + } if txid == confirmed_txid + ), + "the confirmed txid, confirmation state, and classification must all be in place", + ); + assert_eq!(record.amount_msat, Some(1_000_000)); + assert_eq!(record.fee_paid_msat, Some(500)); + } + #[tokio::test] async fn funding_classification_update_or_insert_preserves_advanced_record() { use bitcoin::hashes::Hash; @@ -1308,7 +1381,8 @@ mod tests { // `update_or_insert` applies only the narrow reclassification when a record exists — no // matter when it appeared — setting the `tx_type` while preserving the confirmation - // state wallet sync owns and the confirmed candidate's figures. + // state wallet sync owns. The update names the confirmed txid, so its + // contribution-derived figures replace the record's wallet-view ones. let store = new_store(vec![advanced.clone()]); let update = PaymentDetailsUpdate::funding_reclassification(fresh.clone()); assert_eq!( @@ -1325,8 +1399,8 @@ mod tests { .. } )); - assert_eq!(merged.amount_msat, Some(2_000_000)); - assert_eq!(merged.fee_paid_msat, Some(999)); + assert_eq!(merged.amount_msat, Some(1_000_000)); + assert_eq!(merged.fee_paid_msat, Some(500)); // And it inserts the fresh details when no record exists yet. let store = new_store(Vec::new()); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 9e5039bfdb..a4c1268fd1 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1405,10 +1405,14 @@ impl Wallet { // the write, and a full merge of the fresh Pending/Unconfirmed details would then // downgrade the confirmation state the wallet-sync events own. `update_or_insert` holds // the store's mutation lock across the whole decision: when a record exists — no matter - // when it appeared — only the classification (`tx_type`) and, while the record is still - // unconfirmed, the broadcast txid and our contribution figures are merged; otherwise the - // fresh details are inserted. - let update = PaymentDetailsUpdate::funding_reclassification(details.clone()); + // when it appeared — only the classification (`tx_type`) and the figures of whichever + // candidate the record's state makes authoritative are merged; otherwise the fresh + // details are inserted. + let update = funding_reclassification_update( + details.clone(), + &candidates, + self.payment_store.get(&details.id).as_ref(), + ); let pending_update = PendingPaymentDetailsUpdate { id: update.id, payment_update: Some(update.clone()), @@ -2187,3 +2191,136 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { .saturating_sub(EMPTY_SCRIPT_SIG_WEIGHT + EMPTY_WITNESS_COUNT_WEIGHT), ) } + +/// Builds the payment-store update for a freshly classified funding payment. `details` describes +/// the actively broadcast candidate, but when the record already confirmed a *different* +/// candidate — wallet sync saw it win before this classification ran — the update instead carries +/// the confirmed candidate's txid and figures from the candidate history, mirroring what +/// [`Wallet::apply_funding_status_update`] reports when confirmation arrives after classification. +/// +/// `current` is an unlocked snapshot; that is safe because [`PaymentDetails::update`] only lets +/// figures onto a confirmed record when the update names the confirmed txid, so an update built +/// against a stale snapshot cannot misapply figures if the record's confirmation moves before the +/// update lands. +fn funding_reclassification_update( + details: PaymentDetails, candidates: &[FundingTxCandidate], current: Option<&PaymentDetails>, +) -> PaymentDetailsUpdate { + let mut update = PaymentDetailsUpdate::funding_reclassification(details); + if let Some(PaymentKind::Onchain { + txid: confirmed_txid, + status: ConfirmationStatus::Confirmed { .. }, + .. + }) = current.map(|payment| &payment.kind) + { + if update.txid != Some(*confirmed_txid) { + if let Some(candidate) = candidates.iter().find(|c| c.txid == *confirmed_txid) { + update.txid = Some(candidate.txid); + update.amount_msat = Some(candidate.amount_msat); + update.fee_paid_msat = Some(candidate.fee_paid_msat); + } + } + } + update +} + +#[cfg(test)] +mod tests { + use bitcoin::hashes::Hash; + + use super::*; + + fn onchain_details(txid: Txid, status: ConfirmationStatus) -> PaymentDetails { + PaymentDetails::new( + PaymentId([42u8; 32]), + PaymentKind::Onchain { txid, status, tx_type: None }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + PaymentStatus::Pending, + ) + } + + fn confirmed_status() -> ConfirmationStatus { + ConfirmationStatus::Confirmed { + block_hash: bitcoin::BlockHash::from_byte_array([8u8; 32]), + height: 100, + timestamp: 1, + } + } + + #[test] + fn funding_reclassification_update_substitutes_the_confirmed_candidate() { + let confirmed_txid = Txid::from_byte_array([1u8; 32]); + let active_txid = Txid::from_byte_array([2u8; 32]); + let candidates = vec![ + FundingTxCandidate { + txid: confirmed_txid, + amount_msat: Some(2_000_000), + fee_paid_msat: Some(999), + }, + FundingTxCandidate { + txid: active_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + ]; + let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed); + + // The record confirmed an earlier candidate: the update reports that candidate, not the + // active one. + let current = onchain_details(confirmed_txid, confirmed_status()); + let update = funding_reclassification_update(details.clone(), &candidates, Some(¤t)); + assert_eq!(update.txid, Some(confirmed_txid)); + assert_eq!(update.amount_msat, Some(Some(2_000_000))); + assert_eq!(update.fee_paid_msat, Some(Some(999))); + + // A confirmed candidate we did not contribute to still substitutes, with empty figures — + // the same figures a confirmation arriving after classification would report. + let uncontributed = vec![FundingTxCandidate { + txid: confirmed_txid, + amount_msat: None, + fee_paid_msat: None, + }]; + let update = + funding_reclassification_update(details.clone(), &uncontributed, Some(¤t)); + assert_eq!(update.txid, Some(confirmed_txid)); + assert_eq!(update.amount_msat, Some(None)); + assert_eq!(update.fee_paid_msat, Some(None)); + } + + #[test] + fn funding_reclassification_update_keeps_the_active_candidate() { + let active_txid = Txid::from_byte_array([2u8; 32]); + let candidates = vec![FundingTxCandidate { + txid: active_txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = onchain_details(active_txid, ConfirmationStatus::Unconfirmed); + + // No record yet: the update describes the active candidate. + let update = funding_reclassification_update(details.clone(), &candidates, None); + assert_eq!(update.txid, Some(active_txid)); + assert_eq!(update.amount_msat, Some(Some(1_000_000))); + + // An unconfirmed record: still the active candidate (RBF rotation). + let unconfirmed = + onchain_details(Txid::from_byte_array([1u8; 32]), ConfirmationStatus::Unconfirmed); + let update = + funding_reclassification_update(details.clone(), &candidates, Some(&unconfirmed)); + assert_eq!(update.txid, Some(active_txid)); + + // The record confirmed the active candidate itself: nothing to substitute. + let current = onchain_details(active_txid, confirmed_status()); + let update = funding_reclassification_update(details.clone(), &candidates, Some(¤t)); + assert_eq!(update.txid, Some(active_txid)); + assert_eq!(update.amount_msat, Some(Some(1_000_000))); + + // A confirmed txid outside the candidate history (e.g. the record is an unrelated + // same-id payment): fall back to the active candidate; `PaymentDetails::update` keeps + // the confirmed figures in place on mismatch. + let foreign = onchain_details(Txid::from_byte_array([9u8; 32]), confirmed_status()); + let update = funding_reclassification_update(details, &candidates, Some(&foreign)); + assert_eq!(update.txid, Some(active_txid)); + } +} From 6168bb019cdfcec8ae4b95e45dd9cde0eacc2234 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 30 Jul 2026 10:52:46 -0500 Subject: [PATCH 05/13] f - Recreate a missing pending index while the payment is still Pending Classification treated an absent pending entry as proof the payment had graduated and only merged into existing entries. But a crash or failed write between the payment-store and pending-store writes leaves a Pending record with no index entry, and that state was never repaired: the payment could no longer graduate (graduation iterates the pending store) and its candidate txids could no longer be mapped back to the record, which for an RBF splice invites a duplicate generic payment. Decide by the post-write payment status instead: while the record is still Pending, insert the missing entry (embedding the post-write record, so a confirmation wallet sync already mirrored keeps driving graduation); once it advanced beyond Pending, keep treating absence as graduated. A graduated payment is never Pending, so the no-reindex rule is preserved by the status gate itself. The repaired state is not constructible in a test: it requires a failure injected between the two store writes, and no such seam exists. The store primitives the decision rests on are unit-tested. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 40 +++++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index a4c1268fd1..ca4674d565 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,7 +54,6 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; -use crate::data_store::DataStoreUpdateOrInsertResult; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -1419,24 +1418,27 @@ impl Wallet { conflicting_txids: None, candidates: candidates.clone(), }; - match self.payment_store.update_or_insert(update, details.clone()).await? { - DataStoreUpdateOrInsertResult::Inserted => { - // First time we record this funding payment: index it for graduation. Wallet sync - // can still land between the payment-store write above and this one and mirror an - // advanced confirmation into the pending store, so this write makes the same - // atomic decision: merge narrowly into an entry that appeared, insert the fresh - // one otherwise. - let pending = PendingPaymentDetails::new(details, Vec::new(), candidates); - self.pending_payment_store.update_or_insert(pending_update, pending).await?; - }, - DataStoreUpdateOrInsertResult::Updated | DataStoreUpdateOrInsertResult::Unchanged => { - // An earlier candidate or a racing wallet sync already recorded this payment. - // `update` is a no-op when the pending entry is absent, so the index is not - // re-created for a payment the graduation path already removed. (A graduated - // payment always has a payment-store record, so it cannot take the `Inserted` - // branch above and be re-indexed.) - self.pending_payment_store.update(pending_update).await?; - }, + self.payment_store.update_or_insert(update, details.clone()).await?; + + // The pending index must exist exactly while the authoritative record is Pending: + // graduation and rebroadcast read it, and a graduated payment must not be re-indexed. + // Deciding by the post-write status rather than by whether the write inserted also + // repairs a missing index — a crash or failed write between the two stores leaves a + // Pending record with no entry, and a merge alone would never recreate it, leaving the + // payment unable to graduate and its txids unmapped. + let recorded = self.payment_store.get(&details.id).unwrap_or(details); + if recorded.status == PaymentStatus::Pending { + // Wallet sync can still land between the payment-store write above and this one and + // mirror an advanced confirmation into the pending store, so this write makes the + // same atomic decision: merge narrowly into an entry that appeared, insert otherwise. + // The inserted entry embeds the post-write record rather than the fresh details, so a + // confirmation wallet sync already recorded keeps driving graduation. + let pending = PendingPaymentDetails::new(recorded, Vec::new(), candidates); + self.pending_payment_store.update_or_insert(pending_update, pending).await?; + } else { + // The payment already advanced beyond Pending: the graduation path removed the + // entry, and `update`'s no-op on absence must not re-create it. + self.pending_payment_store.update(pending_update).await?; } Ok(()) } From da0d838f9c50f42b05e0802dff4bd48c3da96e40 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Mon, 3 Aug 2026 10:28:29 -0500 Subject: [PATCH 06/13] f - Read the pending-index gate's status inside the pending store's lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The check that only Pending payments enter the pending index read the payment store before taking the pending store's lock. Graduation could write Succeeded and remove the index entry between the check and the write, and the stale check would then re-create an entry for the graduated payment. The next chain tip re-graduates it from the entry's stale embedded copy — or, if that copy was still unconfirmed, keeps rebroadcasting an already-confirmed transaction on every tip. Move the decision into the pending store's critical section via a new DataStore::mutate that reads, transforms, and persists an entry under one hold of the mutation lock. Re-reading the payment's status there is race-free because graduation writes Succeeded before removing the entry: a read that still observes Pending precedes the removal, which then also deletes anything inserted here. The closure runs with the in-memory map lock released, so it may read other stores without ordering the stores' map locks against each other. The race spans a few instructions between two store writes and no seam exists to schedule a graduation inside it, so no test exercises it; the new unit tests cover the mutate primitive itself. Generated with assistance from Claude Code. Co-Authored-By: Claude Fable 5 --- src/data_store.rs | 155 ++++++++++++++++++++++++++++++++++++++++++++++ src/wallet/mod.rs | 61 +++++++++++------- 2 files changed, 195 insertions(+), 21 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index b7748b172e..a3affbb461 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -223,6 +223,36 @@ where Ok(result) } + /// Atomically transforms the entry for `id` through `f` and persists the result. + /// + /// `f` receives the current entry (`None` when absent) and returns the new state to write; + /// returning `None` leaves the store untouched. The read, the closure, and the write share + /// one critical section of the mutation lock, so no concurrent writer can land in between — + /// unlike a separate [`Self::get`] followed by an insert or update. + /// + /// The closure runs on a clone of the entry with the in-memory map lock released, so it may + /// freely read this store or others (reads see the pre-mutation state) without ordering map + /// locks against each other. Keep it cheap and non-blocking. + /// + /// Returns the written object, or `None` when the closure declined to write. + pub(crate) async fn mutate) -> Option>( + &self, id: &SO::Id, f: F, + ) -> Result, Error> { + let _guard = self.mutation_lock.lock().await; + + let current = self.objects.lock().expect("lock").get(id).cloned(); + let new_object = match f(current.as_ref()) { + Some(new_object) => new_object, + None => return Ok(None), + }; + debug_assert!(new_object.id() == *id, "mutate closure must not change the object's id"); + + self.persist(&new_object).await?; + let mut locked_objects = self.objects.lock().expect("lock"); + locked_objects.insert(new_object.id(), new_object.clone()); + Ok(Some(new_object)) + } + /// Returns in-memory objects matching `f`. /// /// The async mutation lock serializes writers, but this synchronous reader cannot wait on it. @@ -552,6 +582,131 @@ mod tests { assert!(data_store.get(&new_id).is_none()); } + #[tokio::test] + async fn mutate_inserts_when_absent() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let primary_namespace = "datastore_test_primary".to_string(); + let secondary_namespace = "datastore_test_secondary".to_string(); + let data_store: DataStore> = DataStore::new( + Vec::new(), + primary_namespace.clone(), + secondary_namespace.clone(), + Arc::clone(&store), + logger, + ); + + let id = TestObjectId { id: [42u8; 4] }; + let object = TestObject { id, data: [23u8; 3] }; + let result = data_store + .mutate(&id, |existing| { + assert!(existing.is_none()); + Some(object) + }) + .await; + assert_eq!(Ok(Some(object)), result); + + assert_eq!(Some(object), data_store.get(&id)); + let store_key = id.encode_to_hex_str(); + assert!(KVStore::read(&*store, &primary_namespace, &secondary_namespace, &store_key) + .await + .is_ok()); + } + + #[tokio::test] + async fn mutate_transforms_existing_entry() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store: DataStore> = DataStore::new( + vec![existing_object], + "datastore_test_primary".to_string(), + "datastore_test_secondary".to_string(), + store, + logger, + ); + + // The closure sees the current entry and derives the new state from it. + let result = data_store + .mutate(&id, |existing| { + let mut new_object = *existing.unwrap(); + new_object.data[0] += 1; + Some(new_object) + }) + .await; + let expected = TestObject { id, data: [24u8, 23u8, 23u8] }; + assert_eq!(Ok(Some(expected)), result); + assert_eq!(Some(expected), data_store.get(&id)); + } + + #[tokio::test] + async fn mutate_runs_the_closure_without_the_map_lock() { + 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, + ); + + // Closures gate cross-store decisions on reads of other stores, which lock their own + // in-memory maps. Holding this store's map lock across the closure would order it + // before theirs and invite lock-order inversions, so the closure must run with the map + // lock released. + let result = data_store + .mutate(&id, |existing| { + assert_eq!(Some(&existing_object), existing); + assert!(data_store.objects.try_lock().is_ok()); + None + }) + .await; + assert_eq!(Ok(None), result); + } + + #[tokio::test] + async fn mutate_persists_nothing_when_closure_declines() { + let id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + // Returning `None` must not attempt a write (the store fails all writes) nor touch memory. + let result = data_store + .mutate(&id, |existing| { + assert_eq!(Some(&existing_object), existing); + None + }) + .await; + assert_eq!(Ok(None), result); + assert_eq!(Some(existing_object), data_store.get(&id)); + } + + #[tokio::test] + async fn mutate_does_not_mutate_memory_if_persist_fails() { + let existing_id = TestObjectId { id: [42u8; 4] }; + let existing_object = TestObject { id: existing_id, data: [23u8; 3] }; + let data_store = new_failing_data_store(vec![existing_object]); + + let changed = TestObject { id: existing_id, data: [24u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.mutate(&existing_id, |_| Some(changed)).await + ); + assert_eq!(Some(existing_object), data_store.get(&existing_id)); + + let new_id = TestObjectId { id: [55u8; 4] }; + let new_object = TestObject { id: new_id, data: [34u8; 3] }; + assert_eq!( + Err(Error::PersistenceFailed), + data_store.mutate(&new_id, |_| Some(new_object)).await + ); + assert!(data_store.get(&new_id).is_none()); + } + #[tokio::test] async fn insert_or_update_does_not_mutate_memory_if_persist_fails() { let existing_id = TestObjectId { id: [42u8; 4] }; diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index ca4674d565..dc14b290e1 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; +use crate::data_store::StorableObject; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; @@ -1412,13 +1413,8 @@ impl Wallet { &candidates, self.payment_store.get(&details.id).as_ref(), ); - let pending_update = PendingPaymentDetailsUpdate { - id: update.id, - payment_update: Some(update.clone()), - conflicting_txids: None, - candidates: candidates.clone(), - }; - self.payment_store.update_or_insert(update, details.clone()).await?; + let id = update.id; + self.payment_store.update_or_insert(update.clone(), 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. @@ -1426,20 +1422,43 @@ impl Wallet { // 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?; - } + // + // The status must be read inside the pending store's critical section. Graduation writes + // `Succeeded` before removing the entry, so a read there that still observes `Pending` + // is ordered before the removal, which then also deletes anything inserted here. A + // status read taken before this write goes stale when graduation lands in between, and + // would re-index the graduated payment. + self.pending_payment_store + .mutate(&id, |existing| { + // The record was written above and payment records are never removed, so absence + // means the write failed out; fall back to the fresh details. + let recorded = self.payment_store.get(&id).unwrap_or(details); + match existing { + // The inserted entry embeds the post-write record rather than the fresh + // details, so a confirmation wallet sync already recorded keeps driving + // graduation. + None if recorded.status == PaymentStatus::Pending => { + Some(PendingPaymentDetails::new(recorded, Vec::new(), candidates)) + }, + // The payment already advanced beyond Pending: the graduation path removed + // the entry and it must not be re-created. + None => None, + // Wallet sync can land between the payment-store write above and this one + // and mirror an advanced confirmation into the pending store: merge only the + // classification into the entry that appeared. + Some(entry) => { + let pending_update = PendingPaymentDetailsUpdate { + id, + payment_update: Some(update), + conflicting_txids: None, + candidates, + }; + let mut updated = entry.clone(); + updated.update(pending_update).then_some(updated) + }, + } + }) + .await?; Ok(()) } From aae9638bb58d2e9548e18eacea04b847a7e1ea68 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 5 Aug 2026 11:19:10 -0500 Subject: [PATCH 07/13] f - Select the reclassified candidate inside the payment store's lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit persist_funding_payment chose which candidate's figures to merge by reading the payment record before taking the store's mutation lock. A wallet sync confirming a candidate between that read and the write left the choice stale: the update still named the actively-broadcast candidate, so the confirmed-figures guard rightly refused it, and the record kept figures no classification derived — wrong for a shared funding output, and frozen permanently if the payment graduated before another event for the confirmed candidate arrived. The whole decision — insert or merge, and which candidate's figures the record's state makes authoritative — now runs inside the store's critical section, where a concurrent confirmation is either fully visible and substituted, or lands after this write and reads the candidate history itself. The race has no test seam (nothing can interpose between the read and the write), so it is not exercised by a test; the decision's single-threaded behavior is unchanged and remains covered by the existing tests. update_or_insert loses its only caller and is removed. Fixes the first finding in https://github.com/lightningdevkit/ldk-node/pull/962#discussion_r3720014093 Implemented with the assistance of AI tooling. Co-Authored-By: Claude Fable 5 --- src/data_store.rs | 147 ------------------------------------------- src/payment/store.rs | 36 +++++++---- src/wallet/mod.rs | 42 ++++++++----- 3 files changed, 51 insertions(+), 174 deletions(-) diff --git a/src/data_store.rs b/src/data_store.rs index a3affbb461..a440a2e1e8 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -40,13 +40,6 @@ 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, @@ -90,9 +83,6 @@ where /// 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; @@ -182,47 +172,6 @@ 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) - } - /// Atomically transforms the entry for `id` through `f` and persists the result. /// /// `f` receives the current entry (`None` when absent) and returns the new state to write; @@ -486,102 +435,6 @@ 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 mutate_inserts_when_absent() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); diff --git a/src/payment/store.rs b/src/payment/store.rs index 8eb787286c..e0f3f93a86 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -1307,13 +1307,13 @@ mod tests { } #[tokio::test] - async fn funding_classification_update_or_insert_preserves_advanced_record() { + async fn funding_classification_merge_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::data_store::DataStore; use crate::io::test_utils::InMemoryStore; use crate::types::{DynStore, DynStoreWrapper}; @@ -1379,16 +1379,22 @@ mod tests { "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 — setting the `tx_type` while preserving the confirmation + // Classification instead 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. 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!( - Ok(DataStoreUpdateOrInsertResult::Updated), - store.update_or_insert(update, fresh.clone()).await - ); + let written = store + .mutate(&id, |existing| match existing { + Some(current) => { + let mut updated = current.clone(); + updated.update(update).then_some(updated) + }, + None => Some(fresh.clone()), + }) + .await; + assert!(matches!(written, Ok(Some(_))), "the reclassification must merge"); let merged = store.get(&id).unwrap(); assert_eq!(merged.status, PaymentStatus::Succeeded); assert!(matches!( @@ -1405,10 +1411,16 @@ mod tests { // 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 written = store + .mutate(&id, |existing| match existing { + Some(current) => { + let mut updated = current.clone(); + updated.update(update).then_some(updated) + }, + None => Some(fresh.clone()), + }) + .await; + assert!(matches!(written, Ok(Some(_))), "the fresh details must insert"); let inserted = store.get(&id).unwrap(); assert_eq!(inserted.status, PaymentStatus::Pending); assert!(matches!( diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index dc14b290e1..905a2b4970 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1400,21 +1400,33 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { - // 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`) 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 id = update.id; - self.payment_store.update_or_insert(update.clone(), details.clone()).await?; + // Everything this write does depends on the record's current state, so all of it must be + // decided inside the store's critical section. When a record exists — no matter when it + // appeared — only the classification (`tx_type`) and the figures of whichever candidate + // the record's state makes authoritative are merged: a full merge of the fresh + // Pending/Unconfirmed details would downgrade the confirmation state the wallet-sync + // events own. Which candidate is authoritative is equally stateful: substituting the + // confirmed candidate's figures requires seeing the confirmation. Selected from a read + // taken before the lock, the choice goes stale when a confirmation lands in between — + // the update still names the actively-broadcast candidate, the confirmed-figures guard + // then rightly refuses it, and the record is left with figures no classification derived. + let id = details.id; + let mut update = None; + self.payment_store + .mutate(&id, |existing| { + let reclassification = + funding_reclassification_update(details.clone(), &candidates, existing); + update = Some(reclassification.clone()); + match existing { + None => Some(details.clone()), + Some(current) => { + let mut updated = current.clone(); + updated.update(reclassification).then_some(updated) + }, + } + }) + .await?; + let update = update.expect("the mutate closure always runs"); // 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. From 6d0a936947eb64bad933f739d415566d2817cb9f Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 5 Aug 2026 11:21:15 -0500 Subject: [PATCH 08/13] f - Apply funding status updates atomically with the funding-type check apply_funding_status_update fetched the payment record, checked that it was a classified funding payment, merged in the new confirmation status, and wrote the result back as separate store operations. A classification racing in between -- merging tx_type and the contribution-derived figures into the record -- would be overwritten by the stale snapshot. Perform the check and the merge under the payment store's mutation lock so no write can interleave, and skip persisting when the merge changes nothing. Implemented with the assistance of AI tooling. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 72 ++++++++++++++++++++++++++++++----------------- 1 file changed, 46 insertions(+), 26 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 905a2b4970..c063fea930 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1571,34 +1571,54 @@ impl Wallet { async fn apply_funding_status_update( &self, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus, ) -> Result { - let Some(mut payment) = self.payment_store.get(&payment_id) else { + // The funding-type gate, the candidate lookup, and the write share the store's mutation + // lock: against a separate `get`, a classification merging in between would have its + // `tx_type` and contribution figures clobbered by this stale snapshot. + let mut handled = None; + self.payment_store + .mutate(&payment_id, |existing| { + let payment = existing?; + let tx_type = match &payment.kind { + PaymentKind::Onchain { + tx_type: + tx_type @ Some( + TransactionType::Funding { .. } + | TransactionType::InteractiveFunding { .. }, + ), + .. + } => tx_type.clone(), + _ => return None, + }; + // Report the figures of the candidate that actually confirmed, which need not be + // the last one broadcast (an earlier, lower-fee candidate may win) and may carry + // no figures at all (`None`) for a round we didn't contribute to. (`direction` is + // invariant across a splice's candidates and cannot be changed through the store + // anyway.) + let mut target = payment.clone(); + if let Some(pending) = self.pending_payment_store.get(&payment_id) { + if let Some(candidate) = pending.candidate(event_txid) { + target.amount_msat = candidate.amount_msat; + target.fee_paid_msat = candidate.fee_paid_msat; + } + } + target.kind = + PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type }; + + // Merge through the update machinery so its rules (e.g. which fields a merge may + // touch) keep applying, and skip the write when nothing changed. + let mut merged = payment.clone(); + if merged.update(target.to_update()) { + handled = Some(merged.clone()); + Some(merged) + } else { + handled = Some(payment.clone()); + None + } + }) + .await?; + let Some(payment) = handled else { return Ok(false); }; - let tx_type = match &payment.kind { - PaymentKind::Onchain { - tx_type: - tx_type @ Some( - TransactionType::Funding { .. } - | TransactionType::InteractiveFunding { .. }, - ), - .. - } => tx_type.clone(), - _ => return Ok(false), - }; - // Report the figures of the candidate that actually confirmed, which need not be the last - // one broadcast (an earlier, lower-fee candidate may win) and may carry no figures at all - // (`None`) for a round we didn't contribute to. (`direction` is invariant across a splice's - // candidates and cannot be changed through the store anyway.) - if let Some(pending) = self.pending_payment_store.get(&payment_id) { - if let Some(candidate) = pending.candidate(event_txid) { - payment.amount_msat = candidate.amount_msat; - payment.fee_paid_msat = candidate.fee_paid_msat; - } - } - - payment.kind = - PaymentKind::Onchain { txid: event_txid, status: confirmation_status, tx_type }; - self.payment_store.insert_or_update(payment.clone()).await?; // Mirror the refreshed confirmation status onto the pending entry: `ChainTipChanged` // graduates by reading the pending entry's details, so it must see the new status. This is // the same dual-write the default `TxConfirmed` path performs; an empty conflicting-txids From db76891b60bfe78241b21372f24d1001750638aa Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Wed, 5 Aug 2026 11:22:27 -0500 Subject: [PATCH 09/13] f - Keep the candidate history consistent with the classified record Classification writes the payment record and the pending entry carrying the candidate history as two store operations. A funding confirmation processed between them saw the record already classified but the candidate history still absent, so it stamped the confirmed candidate's txid with a stale snapshot's figures -- and once the candidates landed, nothing revisited the payment record to repair them, leaving the wrong amount/fee to graduate with the payment. A wallet-level lock now serializes classification's two-store write pair against the funding-confirmation handling, so a confirmation either runs before the record is classified or sees the full candidate history. Writers touching a single store (e.g. graduation) are unaffected; the per-store gates continue to cover them. The race needs a confirmation interposed between two writes of one classification call, which no test can arrange; the lock is uncontended in the single-writer paths the existing tests exercise. Fixes the finding in https://github.com/lightningdevkit/ldk-node/pull/962#discussion_r3720054363 Implemented with the assistance of AI tooling. Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index c063fea930..7c1e155d9b 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -95,6 +95,14 @@ pub(crate) struct Wallet { config: Arc, logger: Arc, pending_payment_store: Arc, + // Serializes the writers that must observe the payment record and its pending-store entry + // (candidate history included) as one consistent unit: classification's two-store write pair + // and wallet sync's funding-confirmation handling. Without it, a confirmation landing between + // classification's payment-store write and its pending-store write sees the record classified + // but the candidate history absent, and stamps the confirmed candidate with another + // candidate's figures — which nothing afterwards repairs. Writers that touch only one store + // (e.g. graduation) stay safe through the per-store gates instead and need not take this. + funding_payment_update_lock: tokio::sync::Mutex<()>, } impl Wallet { @@ -118,6 +126,7 @@ impl Wallet { config, logger, pending_payment_store, + funding_payment_update_lock: tokio::sync::Mutex::new(()), } } @@ -1400,6 +1409,10 @@ impl Wallet { async fn persist_funding_payment( &self, details: PaymentDetails, candidates: Vec, ) -> Result<(), Error> { + // Hold the cross-store lock across both writes so a funding confirmation never observes + // the record classified but the candidate history it needs still missing. + let _guard = self.funding_payment_update_lock.lock().await; + // Everything this write does depends on the record's current state, so all of it must be // decided inside the store's critical section. When a record exists — no matter when it // appeared — only the classification (`tx_type`) and the figures of whichever candidate @@ -1571,6 +1584,11 @@ impl Wallet { async fn apply_funding_status_update( &self, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus, ) -> Result { + // The cross-store lock orders this against classification's two-store write pair: the + // candidate whose figures are reported below is only reliable once the classification + // that recorded the candidate history has fully landed. + let _guard = self.funding_payment_update_lock.lock().await; + // The funding-type gate, the candidate lookup, and the write share the store's mutation // lock: against a separate `get`, a classification merging in between would have its // `tx_type` and contribution figures clobbered by this stale snapshot. From e21f10a35016ca22e95162a0820b3c84e0721efc Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 6 Aug 2026 11:01:20 -0500 Subject: [PATCH 10/13] f - Serialize wallet sync's payment writes with classification The cross-store lock was drawn too narrowly: wallet sync's event arms resolved the payment id before acquiring it and ran their generic fallback after releasing it. A classification landing inside that window made sync resolve the id against a torn candidate index (minting a duplicate record keyed by the event txid) or overwrite the freshly classified record's contribution figures with wallet-derived ones. Each sync arm now holds the lock from payment-id resolution through its last write. That includes TxReplaced, which was not named in review but has the same boundary hole: the pending entry it writes embeds a read of the payment record that could go stale against classification. apply_funding_status_update no longer acquires the lock itself; it is renamed with a _locked suffix and takes a guard reference to remind callers of the contract, since all of its callers now hold the lock across a wider span than the call. Two barrier tests pin both orderings by parking one writer inside its critical section and dispatching the other: a confirmation arriving during classification's torn window must wait rather than duplicate, and a classification arriving during sync's fallback window must wait rather than be overwritten. Found by joostjager and Codex. This commit was authored with AI assistance (Claude Code). Co-Authored-By: Claude Fable 5 --- src/payment/pending_payment_store.rs | 2 +- src/wallet/mod.rs | 399 +++++++++++++++++++++++++-- 2 files changed, 380 insertions(+), 21 deletions(-) diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index a5994c8808..30a1135374 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -253,7 +253,7 @@ mod tests { 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. + // `apply_funding_status_update_locked`) before classification ran. let confirmed_details = PaymentDetails::new( payment_id, PaymentKind::Onchain { diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 7c1e155d9b..f24f776440 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -96,12 +96,13 @@ pub(crate) struct Wallet { logger: Arc, pending_payment_store: Arc, // Serializes the writers that must observe the payment record and its pending-store entry - // (candidate history included) as one consistent unit: classification's two-store write pair - // and wallet sync's funding-confirmation handling. Without it, a confirmation landing between - // classification's payment-store write and its pending-store write sees the record classified - // but the candidate history absent, and stamps the confirmed candidate with another - // candidate's figures — which nothing afterwards repairs. Writers that touch only one store - // (e.g. graduation) stay safe through the per-store gates instead and need not take this. + // (candidate history included) as one consistent unit: classification holds it across its + // two-store write pair, and wallet sync's event arms hold it from payment-id resolution + // through their last write. Without it, a confirmation landing between classification's two + // writes sees the record classified but the candidate history absent — resolving the wrong + // payment id or stamping the confirmed candidate with another candidate's figures — and a + // classification landing inside an arm's decision sequence gets overwritten by the arm's + // stale generic fallback. funding_payment_update_lock: tokio::sync::Mutex<()>, } @@ -266,12 +267,23 @@ impl Wallet { timestamp: block_time.confirmation_time, }; + // Hold the cross-store lock from payment-id resolution through the last write: + // a classification landing in between would leave the id resolved against a + // torn candidate index and the generic fallback below overwriting (or + // duplicating) the record classification just wrote. + let guard = self.funding_payment_update_lock.lock().await; + let payment_id = self .find_payment_by_txid(txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self - .apply_funding_status_update(payment_id, txid, confirmation_status) + .apply_funding_status_update_locked( + &guard, + payment_id, + txid, + confirmation_status, + ) .await? { continue; @@ -363,12 +375,17 @@ impl Wallet { } }, WalletEvent::TxUnconfirmed { txid, tx, .. } => { + // See `TxConfirmed`: id resolution and the writes below must not interleave + // with classification. + let guard = self.funding_payment_update_lock.lock().await; + let payment_id = self .find_payment_by_txid(txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self - .apply_funding_status_update( + .apply_funding_status_update_locked( + &guard, payment_id, txid, ConfirmationStatus::Unconfirmed, @@ -395,6 +412,12 @@ impl Wallet { self.pending_payment_store.insert_or_update(pending_payment).await?; }, WalletEvent::TxReplaced { txid, conflicts, .. } => { + // See `TxConfirmed`: id resolution and the writes below must not interleave + // with classification. The pending entry written below embeds a read of the + // payment record, which must not go stale against a concurrent + // classification either. + let _guard = self.funding_payment_update_lock.lock().await; + let Some(payment_id) = self.find_payment_by_txid(txid) else { log_error!( self.logger, @@ -425,12 +448,17 @@ impl Wallet { self.pending_payment_store.insert_or_update(pending_payment_details).await?; }, WalletEvent::TxDropped { txid, tx } => { + // See `TxConfirmed`: id resolution and the writes below must not interleave + // with classification. + let guard = self.funding_payment_update_lock.lock().await; + let payment_id = self .find_payment_by_txid(txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self - .apply_funding_status_update( + .apply_funding_status_update_locked( + &guard, payment_id, txid, ConfirmationStatus::Unconfirmed, @@ -1468,9 +1496,10 @@ impl Wallet { // The payment already advanced beyond Pending: the graduation path removed // the entry and it must not be re-created. None => None, - // Wallet sync can land between the payment-store write above and this one - // and mirror an advanced confirmation into the pending store: merge only the - // classification into the entry that appeared. + // The entry predates this classification — wallet sync recorded the + // transaction before it was classified (its arms and this write pair + // serialize on the cross-store lock, so nothing lands in between): merge + // only the classification into the existing entry. Some(entry) => { let pending_update = PendingPaymentDetailsUpdate { id, @@ -1581,14 +1610,15 @@ impl Wallet { /// `sent`/`received` don't capture our contribution to a shared funding output. Returns `true` /// when it handled the payment, so the caller skips the default on-chain path. Graduation to /// `Succeeded` is left to `ChainTipChanged` after `ANTI_REORG_DELAY`. - async fn apply_funding_status_update( - &self, payment_id: PaymentId, event_txid: Txid, confirmation_status: ConfirmationStatus, + /// + /// The caller must hold [`Self::funding_payment_update_lock`] — from resolving `payment_id` + /// through its own last write, not just across this call — so that classification's two-store + /// write pair cannot interleave with the caller's decision sequence. The `_guard` parameter + /// serves as a reminder of that contract. + async fn apply_funding_status_update_locked( + &self, _guard: &tokio::sync::MutexGuard<'_, ()>, payment_id: PaymentId, event_txid: Txid, + confirmation_status: ConfirmationStatus, ) -> Result { - // The cross-store lock orders this against classification's two-store write pair: the - // candidate whose figures are reported below is only reliable once the classification - // that recorded the candidate history has fully landed. - let _guard = self.funding_payment_update_lock.lock().await; - // The funding-type gate, the candidate lookup, and the write share the store's mutation // lock: against a separate `get`, a classification merging in between would have its // `tx_type` and contribution figures clobbered by this stale snapshot. @@ -2267,7 +2297,8 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { /// 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. +/// [`Wallet::apply_funding_status_update_locked`] 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 @@ -2296,9 +2327,193 @@ fn funding_reclassification_update( #[cfg(test)] mod tests { + use std::time::Duration; + + use bdk_chain::{BlockId, ConfirmationBlockTime}; use bitcoin::hashes::Hash; + use bitcoin::Network; + use lightning::io; + use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use super::*; + use crate::config::EsploraSyncConfig; + use crate::io::test_utils::InMemoryStore; + use crate::io::{ + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + }; + use crate::tx_broadcaster::TransactionBroadcaster; + use crate::types::{DynStore, DynStoreWrapper}; + use crate::{NodeMetrics, PersistedNodeMetrics}; + + const EXTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/0/*)"; + const INTERNAL_DESCRIPTOR: &str = "wpkh(tprv8ZgxMBicQKsPdy6LMhUtFHAgpocR8GC6QmwMSFpZs7h6Eziw3SpThFfczTDh5rW2krkqffa11UpX3XkeTTB2FvzZKWXqPY54Y6Rq4AQ5R8L/84'/1'/0'/1/*)"; + + /// A pass-through [`KVStore`] that parks writes to one namespace: a matching writer first + /// signals `parked`, then waits until the test drops its `gate` write guard. Writes to every + /// other namespace pass straight through. + #[derive(Clone)] + struct NamespaceGatedStore { + inner: Arc, + gated_namespace: String, + parked: Arc, + gate: Arc>, + } + + impl NamespaceGatedStore { + fn new(gated_namespace: &str) -> Self { + Self { + inner: Arc::new(InMemoryStore::new()), + gated_namespace: gated_namespace.to_string(), + parked: Arc::new(tokio::sync::Notify::new()), + gate: Arc::new(tokio::sync::RwLock::new(())), + } + } + } + + impl KVStore for NamespaceGatedStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, io::Error>> + 'static + Send { + KVStore::read(&*self.inner, primary_namespace, secondary_namespace, key) + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> impl Future> + 'static + Send { + let inner = Arc::clone(&self.inner); + let gated = primary_namespace == self.gated_namespace; + let parked = Arc::clone(&self.parked); + let gate = Arc::clone(&self.gate); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + async move { + if gated { + parked.notify_one(); + let _guard = gate.read().await; + } + KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf).await + } + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> impl Future> + 'static + Send { + KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy) + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> impl Future, io::Error>> + 'static + Send { + KVStore::list(&*self.inner, primary_namespace, secondary_namespace) + } + } + + impl PaginatedKVStore for NamespaceGatedStore { + fn list_paginated( + &self, primary_namespace: &str, secondary_namespace: &str, + page_token: Option, + ) -> impl Future> + 'static + Send { + PaginatedKVStore::list_paginated( + &*self.inner, + primary_namespace, + secondary_namespace, + page_token, + ) + } + } + + /// Builds a [`Wallet`] whose every component is in-memory; nothing performs I/O at + /// construction, so wallet-sync and classification entry points can be driven directly. + async fn new_test_wallet(kv_store: Arc) -> Wallet { + let logger = Arc::new(Logger::new_log_facade()); + let mut wallet_persister = + KVStoreWalletPersister::new(Arc::clone(&kv_store), Arc::clone(&logger)); + let bdk_wallet = bdk_wallet::Wallet::create(EXTERNAL_DESCRIPTOR, INTERNAL_DESCRIPTOR) + .network(Network::Regtest) + .create_wallet_async(&mut wallet_persister) + .await + .unwrap(); + let broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger))); + let fee_estimator = Arc::new(OnchainFeeEstimator::new()); + let config = Arc::new(Config::default()); + let node_metrics = Arc::new(PersistedNodeMetrics::new(NodeMetrics::default())); + let (chain_source, _) = ChainSource::new_esplora( + "http://127.0.0.1:1".to_string(), + HashMap::new(), + EsploraSyncConfig::default(), + Arc::clone(&fee_estimator), + Arc::clone(&broadcaster), + Arc::clone(&kv_store), + Arc::clone(&config), + Arc::clone(&logger), + node_metrics, + ) + .unwrap(); + let payment_store = Arc::new(PaymentStore::new( + Vec::new(), + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )); + let pending_payment_store = Arc::new(PendingPaymentStore::new( + Vec::new(), + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(), + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )); + let runtime = Arc::new(Runtime::new(Arc::clone(&logger)).unwrap()); + Wallet::new( + bdk_wallet, + wallet_persister, + broadcaster, + fee_estimator, + Arc::new(chain_source), + payment_store, + runtime, + config, + logger, + pending_payment_store, + ) + } + + fn dummy_tx() -> Transaction { + Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: Vec::new(), + output: Vec::new(), + } + } + + fn confirmed_block_time(height: u32) -> ConfirmationBlockTime { + ConfirmationBlockTime { + block_id: BlockId { height, hash: bitcoin::BlockHash::from_byte_array([9u8; 32]) }, + confirmation_time: 100, + } + } + + fn interactive_funding_details( + id: PaymentId, txid: Txid, amount_msat: Option, fee_paid_msat: Option, + ) -> PaymentDetails { + let kind = PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: Some(TransactionType::InteractiveFunding { channels: vec![] }), + }; + PaymentDetails::new( + id, + kind, + amount_msat, + fee_paid_msat, + PaymentDirection::Outbound, + PaymentStatus::Pending, + ) + } fn onchain_details(txid: Txid, status: ConfirmationStatus) -> PaymentDetails { PaymentDetails::new( @@ -2394,4 +2609,148 @@ mod tests { let update = funding_reclassification_update(details, &candidates, Some(&foreign)); assert_eq!(update.txid, Some(active_txid)); } + + /// Barrier test, classification-first ordering: wallet sync's confirmation handling must + /// wait for classification's two-store write pair. Classification is parked between its + /// payment-store and pending-store writes (the torn window) and only then is the + /// confirmation of the replacement candidate dispatched; unless the sync arm holds the + /// cross-store lock from payment-id resolution onwards, it resolves the id against the + /// still-missing pending index and mints a duplicate record keyed by the event txid. + #[tokio::test] + async fn funding_confirmation_waits_for_classification() { + let gated = NamespaceGatedStore::new(PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE); + let store: Arc = Arc::new(DynStoreWrapper(gated.clone())); + let wallet = Arc::new(new_test_wallet(store).await); + + let txid1 = Txid::from_byte_array([1u8; 32]); + let txid2 = Txid::from_byte_array([2u8; 32]); + let payment_id = PaymentId(txid1.to_byte_array()); + let candidates = vec![ + FundingTxCandidate { + txid: txid1, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + FundingTxCandidate { + txid: txid2, + amount_msat: Some(2_000_000), + fee_paid_msat: Some(999), + }, + ]; + let details = interactive_funding_details(payment_id, txid2, Some(2_000_000), Some(999)); + + // Hold the gate so classification parks on its pending-store write: the payment record + // is persisted, the pending entry is not — the torn window a concurrent confirmation + // must not observe. + let gate_guard = gated.gate.write().await; + let classification = tokio::spawn({ + let wallet = Arc::clone(&wallet); + let candidates = candidates.clone(); + async move { wallet.persist_funding_payment(details, candidates).await } + }); + gated.parked.notified().await; + + // Only now dispatch the confirmation of the candidate that won. + let event = WalletEvent::TxConfirmed { + txid: txid2, + tx: Arc::new(dummy_tx()), + block_time: confirmed_block_time(5), + old_block_time: None, + }; + let sync = tokio::spawn({ + let wallet = Arc::clone(&wallet); + async move { wallet.update_payment_store(vec![event]).await } + }); + + // Liveness sanity only (both pre- and post-fix stall here): while classification is + // parked, no second record may have been committed. + tokio::time::sleep(Duration::from_millis(250)).await; + assert!(wallet.payment_store.list_filter(|_| true).len() <= 1); + + drop(gate_guard); + classification.await.unwrap().unwrap(); + sync.await.unwrap().unwrap(); + + // Both writers converge on the classified record: the confirmation refreshes it in + // place with the confirmed candidate's figures rather than minting a second record + // keyed by the event txid. + let payments = wallet.payment_store.list_filter(|_| true); + assert_eq!(payments.len(), 1, "the confirmation must not mint a duplicate record"); + let payment = &payments[0]; + assert_eq!(payment.id, payment_id); + assert_eq!(payment.amount_msat, Some(2_000_000)); + assert_eq!(payment.fee_paid_msat, Some(999)); + match &payment.kind { + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + } => assert_eq!(*txid, txid2), + kind => panic!("unexpected kind {:?}", kind), + } + } + + /// Barrier test, sync-first ordering: classification must wait for wallet sync's complete + /// decision-plus-write sequence. Wallet sync is parked inside its generic-fallback window — + /// past the funding-status check that found no record, before its writes — by holding the + /// BDK wallet lock the fallback needs. Unless the sync arm holds the cross-store lock + /// across that window, classification lands in between and the fallback's stale merge + /// overwrites the contribution-derived figures with wallet-derived ones. + #[tokio::test(flavor = "multi_thread")] + async fn funding_classification_waits_for_wallet_sync() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = Arc::new(new_test_wallet(store).await); + + let txid = Txid::from_byte_array([3u8; 32]); + let payment_id = PaymentId(txid.to_byte_array()); + let candidates = vec![FundingTxCandidate { + txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = interactive_funding_details(payment_id, txid, Some(1_000_000), Some(500)); + + // Park wallet sync inside its fallback window: the TxUnconfirmed arm reads no wallet + // state before that point, so it passes the funding-status check (no record exists yet) + // and then blocks on the wallet lock held here. The sleeps give the tasks time to reach + // their parking spots; they make the pre-fix failure deterministic, while the fixed + // code converges to the same final state under any arrival order. + let inner_guard = wallet.inner.lock().unwrap(); + let sync = tokio::spawn({ + let wallet = Arc::clone(&wallet); + let event = + WalletEvent::TxUnconfirmed { txid, tx: Arc::new(dummy_tx()), old_block_time: None }; + async move { wallet.update_payment_store(vec![event]).await } + }); + tokio::time::sleep(Duration::from_millis(250)).await; + + let classification = tokio::spawn({ + let wallet = Arc::clone(&wallet); + let candidates = candidates.clone(); + async move { wallet.persist_funding_payment(details, candidates).await } + }); + tokio::time::sleep(Duration::from_millis(250)).await; + + drop(inner_guard); + sync.await.unwrap().unwrap(); + classification.await.unwrap().unwrap(); + + // Both writers converge on one record carrying the classification: the generic + // fallback must not clobber the contribution-derived figures with its wallet-derived + // view of the transaction. + let payments = wallet.payment_store.list_filter(|_| true); + assert_eq!(payments.len(), 1); + let payment = &payments[0]; + assert_eq!(payment.id, payment_id); + assert_eq!( + payment.amount_msat, + Some(1_000_000), + "wallet sync's fallback must not overwrite contribution figures" + ); + assert_eq!(payment.fee_paid_msat, Some(500)); + assert!(matches!( + &payment.kind, + PaymentKind::Onchain { tx_type: Some(TransactionType::InteractiveFunding { .. }), .. } + )); + } } From 15dbd99fed7fdcca03e916465be9d3bab82eb540 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 6 Aug 2026 11:03:25 -0500 Subject: [PATCH 11/13] f - Map candidate txids back to the funding record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit find_payment_by_txid resolved a txid through the record's id, its current txid, and its conflicting_txids, but not through the candidate history. A middle candidate from three or more RBF rounds matches none of those — it is not the first candidate (which keys the record), not the active one, and may never have received a TxReplaced event of its own — so wallet sync fell back to a txid-derived id and treated the round as an unrelated payment. Probe the candidate history as well. This also lets TxReplaced resolve such ids; its comment now covers why the payment record is present in that case too. Found by joostjager. This commit was authored with AI assistance (Claude Code). Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 52 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 49 insertions(+), 3 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index f24f776440..34436bdfe1 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -432,9 +432,11 @@ impl Wallet { conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect(); conflict_txids.push(txid); - // The payment already exists in the store at this point: `bump_fee_rbf` updates - // the payment store with the replacement txid before the next sync cycle, so we - // can safely fetch it here. + // The payment already exists in the store at this point: `bump_fee_rbf` + // updates the payment store with the replacement txid before the next sync + // cycle, and an id resolved through the candidate history comes from a + // classification whose payment-store write strictly precedes the candidate + // history it was resolved from. So we can safely fetch it here. debug_assert!( self.payment_store.get(&payment_id).is_some(), "Payment {:?} expected in store during WalletEvent::TxReplaced but not found", @@ -1595,6 +1597,10 @@ impl Wallet { .list_filter(|p| { matches!(p.details.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid) || p.conflicting_txids.contains(&target_txid) + // A middle RBF round is not the record's current txid and may never have + // received a `TxReplaced` event of its own, so map any of its candidate + // txids (an earlier RBF round may confirm) back to the record. + || p.candidate(target_txid).is_some() }) .first() { @@ -2610,6 +2616,46 @@ mod tests { assert_eq!(update.txid, Some(active_txid)); } + /// A middle RBF candidate must map back to the funding record: it is neither the record's + /// id (derived from the first candidate), nor its current txid (the active candidate), nor + /// in `conflicting_txids` (it never got a `TxReplaced` event of its own). + #[tokio::test] + async fn find_payment_by_txid_maps_candidate_txids() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store).await; + + let txid1 = Txid::from_byte_array([1u8; 32]); + let txid2 = Txid::from_byte_array([2u8; 32]); + let txid3 = Txid::from_byte_array([3u8; 32]); + let payment_id = PaymentId(txid1.to_byte_array()); + let candidates = vec![ + FundingTxCandidate { + txid: txid1, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }, + FundingTxCandidate { + txid: txid2, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(600), + }, + FundingTxCandidate { + txid: txid3, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(700), + }, + ]; + let details = interactive_funding_details(payment_id, txid3, Some(1_000_000), Some(700)); + let entry = PendingPaymentDetails::new(details, Vec::new(), candidates); + wallet.pending_payment_store.insert_or_update(entry).await.unwrap(); + + // The first candidate resolves via the txid-derived id and the active candidate via the + // record's current txid; the middle one must resolve through the candidate history. + assert_eq!(wallet.find_payment_by_txid(txid1), Some(payment_id)); + assert_eq!(wallet.find_payment_by_txid(txid3), Some(payment_id)); + assert_eq!(wallet.find_payment_by_txid(txid2), Some(payment_id)); + } + /// Barrier test, classification-first ordering: wallet sync's confirmation handling must /// wait for classification's two-store write pair. Classification is parked between its /// payment-store and pending-store writes (the torn window) and only then is the From 71c565fd249e7b5bc6e75a93cfe58eb943320254 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 6 Aug 2026 11:07:14 -0500 Subject: [PATCH 12/13] f - Graduate payments without rewriting them from a stale snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Graduation wrote the pending-store snapshot back to the payment store with only its status flipped to Succeeded. That write carries the snapshot's figures and confirmation status, so a classification landing between the snapshot and the write had its contribution-derived figures rolled back to wallet-derived ones — the confirmed-figures merge rule does not protect a record against an update naming its own confirmed txid. Graduation now decides from the live record inside the payment store's mutation lock and writes a status-only update: it carries nothing a concurrent classification could lose, bumps the record's update timestamp through the regular update machinery, and no-ops when the record is already Succeeded (the leaked pending entry is still removed). This keeps graduation off the cross-store funding lock — the snapshot's depth check remains a cheap pre-filter, and the closure re-verifies against live state. Deciding from the live record also means a record that diverged from the snapshot declines instead of being force-graduated. That arm is hardening rather than a reachable-bug fix: no current production writer downgrades a record's confirmation behind the snapshot. One behavioral consequence: a record deleted via Node::remove_payment now leaves its pending entry in place (declining each tip change) instead of being resurrected from the snapshot — a zombie index entry over restoring user-deleted data. Found by Codex. This commit was authored with AI assistance (Claude Code). Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 146 ++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 141 insertions(+), 5 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 34436bdfe1..0a86aa3c7b 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -102,7 +102,9 @@ pub(crate) struct Wallet { // writes sees the record classified but the candidate history absent — resolving the wrong // payment id or stamping the confirmed candidate with another candidate's figures — and a // classification landing inside an arm's decision sequence gets overwritten by the arm's - // stale generic fallback. + // stale generic fallback. Graduation stays off this lock: it decides from the live record + // under the payment store's mutation lock and writes only the status, so it carries nothing + // a concurrent classification could lose. funding_payment_update_lock: tokio::sync::Mutex<()>, } @@ -324,7 +326,7 @@ impl Wallet { let mut unconfirmed_outbound_txids: Vec = Vec::new(); - for mut payment in pending_payments { + for payment in pending_payments { match payment.details.kind { PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { height, .. }, @@ -332,9 +334,41 @@ impl Wallet { } => { let payment_id = payment.details.id; if new_tip.height >= height + ANTI_REORG_DELAY - 1 { - payment.details.status = PaymentStatus::Succeeded; - self.payment_store.insert_or_update(payment.details).await?; - self.pending_payment_store.remove(&payment_id).await?; + // Graduate from the live record, not the snapshot listed + // above: a classification landing since then must not have + // its figures rolled back. The status-only update carries + // no figures/txid/confirmation, so nothing a concurrent + // writer wrote can be clobbered; the update machinery bumps + // `latest_update_timestamp` and no-ops when the record is + // already `Succeeded`. A record that has diverged from the + // snapshot (or was removed) declines, leaving future + // events to drive it. + let mut graduated = false; + self.payment_store + .mutate(&payment_id, |existing| { + let current = existing?; + match current.kind { + PaymentKind::Onchain { + status: + ConfirmationStatus::Confirmed { height, .. }, + .. + } if new_tip.height + >= height + ANTI_REORG_DELAY - 1 => + { + graduated = true; + let mut update = + PaymentDetailsUpdate::new(payment_id); + update.status = Some(PaymentStatus::Succeeded); + let mut updated = current.clone(); + updated.update(update).then_some(updated) + }, + _ => None, + } + }) + .await?; + if graduated { + self.pending_payment_store.remove(&payment_id).await?; + } } }, PaymentKind::Onchain { @@ -2616,6 +2650,108 @@ mod tests { assert_eq!(update.txid, Some(active_txid)); } + /// Graduation must decide from the live record and write only the status: a pending-store + /// snapshot taken before a concurrent classification landed must not roll the record's + /// figures back when the payment graduates to `Succeeded`. + #[tokio::test] + async fn graduation_preserves_classified_figures() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store).await; + + let txid = Txid::from_byte_array([4u8; 32]); + let payment_id = PaymentId(txid.to_byte_array()); + let confirmed = ConfirmationStatus::Confirmed { + block_hash: bitcoin::BlockHash::from_byte_array([9u8; 32]), + height: 5, + timestamp: 100, + }; + let tx_type = Some(TransactionType::InteractiveFunding { channels: vec![] }); + + // The live record carries the classification: contribution-derived figures, confirmed. + let mut recorded = + interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999)); + recorded.kind = PaymentKind::Onchain { txid, status: confirmed, tx_type: tx_type.clone() }; + recorded.latest_update_timestamp = 0; + wallet.payment_store.insert_or_update(recorded).await.unwrap(); + + // The pending entry embeds a stale snapshot: wallet-derived figures recorded before the + // classification above landed. + let mut stale = interactive_funding_details(payment_id, txid, Some(0), Some(0)); + stale.kind = PaymentKind::Onchain { txid, status: confirmed, tx_type }; + let entry = PendingPaymentDetails::new(stale, Vec::new(), Vec::new()); + wallet.pending_payment_store.insert_or_update(entry).await.unwrap(); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { old_tip: block_id(9), new_tip: block_id(10) }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).unwrap(); + assert_eq!(payment.status, PaymentStatus::Succeeded); + assert_eq!( + payment.amount_msat, + Some(2_000_000), + "graduation must not roll figures back to the snapshot's" + ); + assert_eq!(payment.fee_paid_msat, Some(999)); + assert!(payment.latest_update_timestamp > 0, "the graduation write must timestamp"); + assert!(wallet.pending_payment_store.get(&payment_id).is_none()); + } + + /// When the live record has diverged from the pending-store snapshot — here the snapshot + /// says Confirmed at graduation depth while the record says Unconfirmed — graduation must + /// decline and keep the entry rather than force-writing `Succeeded` from stale state. The + /// seeded divergence is synthetic (no current production writer downgrades a record's + /// confirmation); the test pins the hardening that comes with deciding from the live record. + #[tokio::test] + async fn graduation_declines_on_diverged_record() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store).await; + + let txid = Txid::from_byte_array([5u8; 32]); + let payment_id = PaymentId(txid.to_byte_array()); + let confirmed = ConfirmationStatus::Confirmed { + block_hash: bitcoin::BlockHash::from_byte_array([9u8; 32]), + height: 5, + timestamp: 100, + }; + + // The live record is Unconfirmed... + let recorded = interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999)); + wallet.payment_store.insert_or_update(recorded).await.unwrap(); + + // ...while the pending entry's snapshot claims a graduation-deep confirmation. + let mut snapshot = + interactive_funding_details(payment_id, txid, Some(2_000_000), Some(999)); + snapshot.kind = PaymentKind::Onchain { + txid, + status: confirmed, + tx_type: Some(TransactionType::InteractiveFunding { channels: vec![] }), + }; + let entry = PendingPaymentDetails::new(snapshot, Vec::new(), Vec::new()); + wallet.pending_payment_store.insert_or_update(entry).await.unwrap(); + + let block_id = + |height| BlockId { height, hash: bitcoin::BlockHash::from_byte_array([7u8; 32]) }; + let event = WalletEvent::ChainTipChanged { old_tip: block_id(9), new_tip: block_id(10) }; + wallet.update_payment_store(vec![event]).await.unwrap(); + + let payment = wallet.payment_store.get(&payment_id).unwrap(); + assert_eq!( + payment.status, + PaymentStatus::Pending, + "a diverged snapshot must not force-graduate the record" + ); + assert!(matches!( + payment.kind, + PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } + )); + assert!( + wallet.pending_payment_store.get(&payment_id).is_some(), + "the entry must survive for future events to drive" + ); + } + /// A middle RBF candidate must map back to the funding record: it is neither the record's /// id (derived from the first candidate), nor its current txid (the active candidate), nor /// in `conflicting_txids` (it never got a `TxReplaced` event of its own). From 6aa35b38f0f5cb2107182a3f5b86ec1b6c220f02 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 6 Aug 2026 11:09:30 -0500 Subject: [PATCH 13/13] f - Correct funding_reclassification_update's locking doc The doc still described current as an unlocked snapshot whose staleness the confirmed-figures merge rule papers over. Since classification moved the candidate choice inside the payment store's mutate closure, current is observed within that critical section and cannot go stale against a concurrent confirmation; the merge rule is a second line of arbitration, not the safety argument. Doc-only change. Found by joostjager. This commit was authored with AI assistance (Claude Code). Co-Authored-By: Claude Fable 5 --- src/wallet/mod.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 0a86aa3c7b..8d9c4e6d23 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -2340,10 +2340,11 @@ fn ldk_to_bdk_satisfaction_weight(ldk_satisfaction_weight: u64) -> Weight { /// [`Wallet::apply_funding_status_update_locked`] 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. +/// `current` is the record as observed inside the payment store's `mutate` critical section — its +/// sole caller, [`Wallet::persist_funding_payment`], builds and applies the update within one +/// closure — so the candidate choice cannot go stale against a concurrent confirmation before the +/// update lands. [`PaymentDetails::update`]'s confirmed-figures rule still arbitrates which +/// figures may land on the record. fn funding_reclassification_update( details: PaymentDetails, candidates: &[FundingTxCandidate], current: Option<&PaymentDetails>, ) -> PaymentDetailsUpdate {