diff --git a/src/data_store.rs b/src/data_store.rs index b1ed816df..b7748b172 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 f5f2fa40a..a5994c880 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -242,4 +242,79 @@ 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 candidates while preserving the + // 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, + 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 d2b92747a..8eb787286 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -243,12 +243,28 @@ 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. 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 { txid, status: ConfirmationStatus::Confirmed { .. }, .. } + if update.txid != Some(txid) + ); + + 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 +294,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); }, _ => {}, @@ -712,6 +728,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 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, 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); + 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 +1065,358 @@ 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 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] + 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)); + } + + #[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; + 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 recorded (unclassified, via the default + // on-chain path) and 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: None, + }, + 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 — 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 merged = store.get(&id).unwrap(); + assert_eq!(merged.status, PaymentStatus::Succeeded); + assert!(matches!( + merged.kind, + 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)); + + // 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 f8d9d521e..ca4674d56 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,47 @@ 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?; + // 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 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?; + + // 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(()) } @@ -2154,3 +2193,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)); + } +}