diff --git a/orange-sdk/src/lib.rs b/orange-sdk/src/lib.rs index 69a99ed..08b2530 100644 --- a/orange-sdk/src/lib.rs +++ b/orange-sdk/src/lib.rs @@ -27,7 +27,7 @@ use ldk_node::lightning::ln::msgs::SocketAddress; use ldk_node::lightning::util::logger::Logger as _; use ldk_node::lightning::{log_debug, log_error, log_info, log_trace, log_warn}; use ldk_node::lightning_invoice::Bolt11Invoice; -use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind}; +use ldk_node::payment::{PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus}; use ldk_node::{BuildError, ChannelDetails, NodeError}; #[cfg(feature = "_test-utils")] pub use lightning_wallet::list_node_payments; @@ -536,15 +536,24 @@ impl From for WalletError { } } -fn should_surface_lightning_payment_without_metadata(status: TxStatus, kind: &PaymentKind) -> bool { - status == TxStatus::Completed || matches!(kind, PaymentKind::Onchain { .. }) +fn should_surface_lightning_payment_without_metadata( + status: TxStatus, kind: &PaymentKind, direction: PaymentDirection, +) -> bool { + // Hide only non-completed *inbound* Lightning records (issued-but-unpaid + // invoices are noise). Outbound attempts always surface: a failed or + // still-pending send the user made must appear in their history — + // otherwise a failed payment leaves no record at all. + status == TxStatus::Completed + || matches!(kind, PaymentKind::Onchain { .. }) + || direction == PaymentDirection::Outbound } fn lightning_payment_without_metadata_to_transaction( payment: &PaymentDetails, fee: Option, ) -> Option { let status = payment.status.into(); - if !should_surface_lightning_payment_without_metadata(status, &payment.kind) { + if !should_surface_lightning_payment_without_metadata(status, &payment.kind, payment.direction) + { return None; } @@ -830,6 +839,9 @@ impl Wallet { } /// Lists the transactions which have been made. + /// Submitted payments include pending and confirmed failed sends across backends. + /// Errors before submission, such as invoice validation or quote preparation failures, + /// do not create history entries. An uncertain send outcome remains pending. pub async fn list_transactions(&self) -> Result, WalletError> { let (trusted_payments, splice_outs) = tokio::join!( self.inner.trusted.list_payments(), @@ -1080,12 +1092,28 @@ impl Wallet { }, } } else { - debug_assert_ne!( - payment.direction, - PaymentDirection::Outbound, - "Missing outbound lightning payment metadata entry on {}", - payment.id - ); + // Only a *successful* outbound payment is expected to have a metadata entry + // (`pay()` writes one whenever the send returns `Ok`). Outbound records can + // legitimately lack metadata when a send fails synchronously — ldk-node + // inserts a `Failed` record before returning `Err`, so `pay()` never + // observes an id — or briefly while a `Pending` record awaits the upsert. + // Even `Succeeded` is only a should-never-happen: a crash between + // ldk-node persisting the payment and the metadata write leaves one. + if payment.direction == PaymentDirection::Outbound + && payment.status == PaymentStatus::Succeeded + { + log_warn!( + self.inner.logger, + "Missing outbound lightning payment metadata entry on {}", + payment.id + ); + #[cfg(feature = "_test-utils")] + debug_assert!( + false, + "Missing outbound lightning payment metadata entry on {}", + payment.id + ); + } if let Some(transaction) = lightning_payment_without_metadata_to_transaction(&payment, fee) @@ -1315,9 +1343,11 @@ impl Wallet { let res = self.inner.trusted.pay(method, instructions.amount).await; match res { Ok(id) => { + // A backend that is already paying this request returns the + // in-flight payment's id, so the entry may already exist. self.inner .tx_metadata - .insert( + .upsert( PaymentId::Trusted(id), TxMetadata { ty: TxType::Payment { ty: ty() }, @@ -1556,6 +1586,20 @@ impl Wallet { ); let payment_hash = invoice.payment_hash(); + + // The lightning leg would be rejected as a duplicate if this invoice is already being + // paid or has been paid. Check before the trusted leg goes out, since that leg cannot be + // recalled and would only fail once the receiver's MPP timeout expires. This is best + // effort; ldk-node still rejects a duplicate the check does not see. + if self + .inner + .ln_wallet + .recent_outbound_bolt11(payment_hash)? + .is_some_and(|p| matches!(p.status, PaymentStatus::Pending | PaymentStatus::Succeeded)) + { + return Err(WalletError::LdkNodeFailure(NodeError::DuplicatePayment)); + } + self.inner.event_queue.begin_mpp_setup(payment_hash).await; // Pay the trusted portion first; the receiver will hold it until the lightning portion @@ -1582,6 +1626,28 @@ impl Wallet { Ok(id) => id, Err(e) => { log_error!(self.inner.logger, "Failed to send lightning MPP portion: {e:?}"); + // Only remove a failed leg. DuplicatePayment refers to an existing payment + // and must never delete its pending or completed history. + let failed_leg = if e == NodeError::DuplicatePayment { + None + } else { + self.inner + .ln_wallet + .recent_outbound_bolt11(payment_hash) + .ok() + .flatten() + .filter(|p| p.status == PaymentStatus::Failed) + }; + if let Some(leg) = failed_leg { + if let Err(remove_err) = + self.inner.ln_wallet.inner.ldk_node.remove_payment(&leg.id) + { + log_error!( + self.inner.logger, + "Failed to remove failed MPP leg: {remove_err:?}" + ); + } + } // The trusted leg is already in flight but there will be no lightning leg to // complete the MPP. Record it as a plain payment so its eventual (failed) terminal // event surfaces normally rather than waiting on a sibling leg that never comes. @@ -1807,7 +1873,11 @@ mod tests { tx_type: None, }; - assert!(should_surface_lightning_payment_without_metadata(TxStatus::Pending, &kind)); + assert!(should_surface_lightning_payment_without_metadata( + TxStatus::Pending, + &kind, + PaymentDirection::Inbound + )); } #[test] @@ -1841,16 +1911,54 @@ mod tests { } #[test] - fn pending_non_onchain_lightning_payments_without_metadata_are_hidden() { + fn pending_inbound_non_onchain_lightning_payments_without_metadata_are_hidden() { let kind = PaymentKind::Spontaneous { hash: PaymentHash([42; 32]), preimage: None }; - assert!(!should_surface_lightning_payment_without_metadata(TxStatus::Pending, &kind)); + assert!(!should_surface_lightning_payment_without_metadata( + TxStatus::Pending, + &kind, + PaymentDirection::Inbound + )); } #[test] fn completed_lightning_payments_without_metadata_are_listed() { let kind = PaymentKind::Spontaneous { hash: PaymentHash([42; 32]), preimage: None }; - assert!(should_surface_lightning_payment_without_metadata(TxStatus::Completed, &kind)); + assert!(should_surface_lightning_payment_without_metadata( + TxStatus::Completed, + &kind, + PaymentDirection::Inbound + )); + } + + #[test] + fn failed_and_pending_outbound_payments_are_listed() { + // A failed or in-flight send the user made must appear in their + // history — a failed payment that leaves no record erodes trust + // in the send flow (the wallet UI can't show what it never sees). + let kind = PaymentKind::Bolt11 { + hash: PaymentHash([42; 32]), + preimage: None, + secret: None, + counterparty_skimmed_fee_msat: None, + }; + + assert!(should_surface_lightning_payment_without_metadata( + TxStatus::Failed, + &kind, + PaymentDirection::Outbound + )); + assert!(should_surface_lightning_payment_without_metadata( + TxStatus::Pending, + &kind, + PaymentDirection::Outbound + )); + // Inbound failures (expired unpaid invoices) stay hidden. + assert!(!should_surface_lightning_payment_without_metadata( + TxStatus::Failed, + &kind, + PaymentDirection::Inbound + )); } } diff --git a/orange-sdk/src/lightning_wallet.rs b/orange-sdk/src/lightning_wallet.rs index d70c82d..b29560c 100644 --- a/orange-sdk/src/lightning_wallet.rs +++ b/orange-sdk/src/lightning_wallet.rs @@ -21,6 +21,7 @@ use ldk_node::lightning::ln::msgs::SocketAddress; use ldk_node::lightning::util::logger::Logger as _; use ldk_node::lightning::{log_debug, log_error, log_info}; use ldk_node::lightning_invoice::{Bolt11Invoice, Bolt11InvoiceDescription, Description}; +use ldk_node::lightning_types::payment::PaymentHash; use ldk_node::payment::{ ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, }; @@ -299,6 +300,21 @@ impl LightningWallet { list_node_payments(&self.inner.ldk_node) } + /// The most recent outbound BOLT 11 record for `payment_hash`, if one was created recently. + /// + /// Only the newest page of the payment list is read, so this neither derives an LDK payment + /// ID from the hash nor walks the history. The failed MPP leg this finds was inserted + /// moments ago, and the duplicate check that also uses it is best effort, since ldk-node + /// rejects a duplicate itself. + pub(crate) fn recent_outbound_bolt11( + &self, payment_hash: PaymentHash, + ) -> Result, NodeError> { + Ok(self.inner.ldk_node.list_payments(None)?.payments.into_iter().find(|p| { + p.direction == PaymentDirection::Outbound + && matches!(p.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash) + })) + } + pub(crate) fn get_balance(&self) -> LightningWalletBalance { let balances = self.inner.ldk_node.list_balances(); LightningWalletBalance { diff --git a/orange-sdk/src/trusted_wallet/cashu/mod.rs b/orange-sdk/src/trusted_wallet/cashu/mod.rs index 75f8f0d..2b21829 100644 --- a/orange-sdk/src/trusted_wallet/cashu/mod.rs +++ b/orange-sdk/src/trusted_wallet/cashu/mod.rs @@ -1,5 +1,6 @@ //! An implementation of `TrustedWalletInterface` using the Cashu (CDK) SDK. +use super::payment_store::PaymentStore; use crate::logging::Logger; use crate::runtime::Runtime; use crate::store::{PaymentId, TxMetadataStore, TxStatus}; @@ -11,7 +12,7 @@ use ldk_node::bitcoin::hashes::Hash; use ldk_node::bitcoin::hashes::sha256::Hash as Sha256; use ldk_node::bitcoin::hex::FromHex; use ldk_node::lightning::util::logger::Logger as _; -use ldk_node::lightning::{log_error, log_info}; +use ldk_node::lightning::{log_debug, log_error, log_info, log_warn}; use ldk_node::lightning_invoice::Bolt11Invoice; use ldk_node::lightning_types::payment::{PaymentHash, PaymentPreimage}; @@ -23,20 +24,21 @@ use cdk::nuts::MeltOptions; use cdk::nuts::nut00::PaymentMethod as CdkPaymentMethod; use cdk::nuts::nut23::Amountless; use cdk::nuts::{CurrencyUnit, MeltQuoteState}; -use cdk::wallet::MintQuote; +use cdk::types::FinalizedMelt; use cdk::wallet::Wallet; use cdk::wallet::types::{Transaction, TransactionDirection}; +use cdk::wallet::{MeltQuote, MintQuote}; use cdk::{Amount as CdkAmount, StreamExt}; use graduated_rebalancer::ReceivedLightningPayment; -use tokio::sync::{mpsc, watch}; +use tokio::sync::{Notify, RwLock, mpsc, watch}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::future::Future; use std::pin::Pin; use std::str::FromStr; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; /// Cashu KV store implementation @@ -58,6 +60,7 @@ pub struct CashuConfig { /// A wallet implementation using the Cashu (CDK) SDK. #[derive(Clone)] pub struct Cashu { + melt: Arc, cashu_wallet: Arc, unit: CurrencyUnit, shutdown_sender: watch::Sender<()>, @@ -65,8 +68,6 @@ pub struct Cashu { supports_bolt12: Arc, supports_mpp: Arc, mint_quote_sender: mpsc::Sender, - event_queue: Arc, - tx_metadata: TxMetadataStore, runtime: Arc, npubcash_url: Option, npub: Option, @@ -184,7 +185,7 @@ impl TrustedWalletInterface for Cashu { .map(|t| Self::convert_transaction_to_payment(t, &self.unit)) .collect::, _>>()?; - Ok(payments) + Ok(self.melt.payments.merge(payments).await) }) } @@ -316,8 +317,7 @@ impl TrustedWalletInterface for Cashu { // We'll use the quote ID as the payment identifier let payment_id = Self::id_to_32_byte_array("e.id); - // Execute the melt in a background task; do not block on it succeeding/failing. - self.spawn_melt(quote.id.clone(), payment_id, payment_hash); + self.start_melt("e, payment_id, amount, payment_hash).await?; Ok(payment_id) }) @@ -354,7 +354,7 @@ impl TrustedWalletInterface for Cashu { })?; let payment_id = Self::id_to_32_byte_array("e.id); - self.spawn_melt(quote.id.clone(), payment_id, payment_hash); + self.start_melt("e, payment_id, partial_amount, payment_hash).await?; Ok(payment_id) }) } @@ -430,6 +430,7 @@ impl Cashu { }, }; + let payments = Arc::new(PaymentStore::new(Arc::clone(&store), Arc::clone(&logger)).await?); let db = Arc::new( CashuKvDatabase::new(Arc::clone(&store), Arc::clone(&runtime)).await.map_err(|e| { InitFailure::TrustedFailure(TrustedError::Other(format!( @@ -477,6 +478,49 @@ impl Cashu { let (shutdown_sender, mut shutdown_receiver) = watch::channel::<()>(()); + let melt = Arc::new(MeltContext { + payments, + in_flight: Mutex::new(HashSet::new()), + gate: RwLock::new(()), + reconcile: Notify::new(), + event_queue: Arc::clone(&event_queue), + tx_metadata: tx_metadata.clone(), + unit: cashu_config.unit.clone(), + logger: Arc::clone(&logger), + }); + + // Resolve melts whose outcome was unknown when the process last stopped, or whose + // request failed in a way that does not prove the mint did not pay. The task wakes up + // when a melt ends without a definite result and backs off while anything is unresolved. + let melt_for_reconcile = Arc::clone(&melt); + let wallet_for_reconcile = Arc::clone(&cashu_wallet); + let mut shutdown_for_reconcile = shutdown_sender.subscribe(); + runtime.spawn_cancellable_background_task(async move { + const MIN_BACKOFF: Duration = Duration::from_secs(30); + const MAX_BACKOFF: Duration = Duration::from_secs(10 * 60); + let mut backoff = MIN_BACKOFF; + loop { + let unresolved = melt_for_reconcile.reconcile(&wallet_for_reconcile).await; + let notified = melt_for_reconcile.reconcile.notified(); + if unresolved == 0 { + backoff = MIN_BACKOFF; + tokio::select! { + _ = shutdown_for_reconcile.changed() => return, + _ = notified => {}, + } + } else { + tokio::select! { + _ = shutdown_for_reconcile.changed() => return, + // A melt just ended without a result; check it promptly. + _ = notified => backoff = MIN_BACKOFF, + _ = tokio::time::sleep(backoff) => { + backoff = (backoff * 2).min(MAX_BACKOFF); + }, + } + } + } + }); + // Create channel for mint quote monitoring with bounded capacity let (mint_quote_sender, mut mint_quote_receiver) = mpsc::channel::(32); @@ -611,6 +655,7 @@ impl Cashu { } Ok(Cashu { + melt, cashu_wallet, unit: cashu_config.unit, shutdown_sender, @@ -618,8 +663,6 @@ impl Cashu { supports_bolt12, supports_mpp, mint_quote_sender, - event_queue, - tx_metadata, runtime, npubcash_url, npub, @@ -642,6 +685,40 @@ impl Cashu { encode::(hrp, &xonly.serialize()).map_err(|e| format!("bech32 encode: {e}")) } + /// Persists the attempt, then melts the quote unless it is already in flight. + /// + /// A quote the mint still lists as unpaid can be melted again after a transient error; the + /// CDK and the mint reject a quote that is actually being paid. A quote that is pending at the + /// mint is left to the reconciliation task, which reports its final outcome. + async fn start_melt( + &self, quote: &MeltQuote, payment_id: [u8; 32], amount: Amount, + payment_hash: Option, + ) -> Result<(), TrustedError> { + if quote.state == MeltQuoteState::Unpaid { + // Register before persisting so reconciliation never sees this payment as abandoned. + if !self.melt.in_flight.lock().unwrap().insert(payment_id) { + log_debug!(self.logger, "Melt for quote {} is already in flight", quote.id); + return Ok(()); + } + let reference = Some(quote.id.clone()); + if let Err(e) = self.melt.payments.insert_pending(payment_id, amount, reference).await { + self.melt.in_flight.lock().unwrap().remove(&payment_id); + return Err(e); + } + self.spawn_melt(quote.id.clone(), payment_id, payment_hash); + } else { + self.melt.payments.insert_pending(payment_id, amount, Some(quote.id.clone())).await?; + log_info!( + self.logger, + "Quote {} is {}; waiting for its outcome instead of melting again", + quote.id, + quote.state + ); + self.melt.reconcile.notify_one(); + } + Ok(()) + } + /// Executes a previously-created melt quote in a background task, emitting a /// [`PaymentSuccessful`] or [`PaymentFailed`] event when it completes. The payment is not /// awaited; this only kicks off the melt. @@ -652,164 +729,28 @@ impl Cashu { &self, quote_id: String, payment_id: [u8; 32], payment_hash: Option, ) { let cashu_wallet = Arc::clone(&self.cashu_wallet); - let logger = Arc::clone(&self.logger); - let event_queue = Arc::clone(&self.event_queue); - let tx_metadata = self.tx_metadata.clone(); - let unit = self.unit.clone(); + let melt = Arc::clone(&self.melt); self.runtime.spawn_background_task(async move { + let gate = melt.gate.read().await; let mut metadata = HashMap::new(); if let Some(hash) = &payment_hash { metadata.insert(PAYMENT_HASH_METADATA_KEY.to_string(), hash.to_string()); } + let mut submitted = false; let melt_result = async { let prepared = cashu_wallet.prepare_melt("e_id, metadata).await?; + submitted = true; prepared.confirm().await } .await; - let fee_paid_msat = melt_result - .as_ref() - .ok() - .and_then(|res| convert_amount(res.fee_paid(), &unit).ok()) - .map(|fee| fee.milli_sats()); - // confirm() waits for a terminal result. Wake registered rebalances even - // when their public payment event is suppressed or cannot be persisted. - if let Some(hash) = payment_hash { - let receipt = melt_result - .as_ref() - .ok() - .filter(|res| res.state() == MeltQuoteState::Paid) - .map(|_| ReceivedLightningPayment { id: payment_id, fee_paid_msat }); - event_queue.rebalance_watchers.sent(hash.0, receipt); - } - match melt_result { - Ok(res) => { - match res.state() { - MeltQuoteState::Paid => { - log_info!(logger, "Successfully sent for quote: {quote_id}"); - - let payment_id = PaymentId::Trusted(payment_id); - let is_rebalance = { - let map = tx_metadata.read(); - map.get(&payment_id).is_some_and(|m| m.ty.is_rebalance()) - }; - if is_rebalance { - return; - } - - let preimage: Option = match res.payment_proof() { - Some(str) => match FromHex::from_hex(str) { - Ok(b) => Some(PaymentPreimage(b)), - Err(e) => { - log_error!( - logger, - "Failed to decode preimage ({:?}) for quote {quote_id}: {e}", - res.payment_proof() - ); - None - }, - }, - None => { - // Expected for same-mint payments: when the melt's - // bolt11 destination is a mint quote on this same - // mint, cdk-mintd settles internally — no Lightning - // payment occurs, so there is no preimage to return. - // The success path below already tolerates None - // (hash falls back to the invoice payment_hash). - log_info!( - logger, - "Melt for quote {quote_id} settled without a preimage (internal/same-mint settlement)" - ); - None - }, - }; - - let hash = match payment_hash { - Some(hash) => hash, - None => { - match preimage { - Some(pre) => { - let hash = Sha256::hash(&pre.0); - PaymentHash(hash.to_byte_array()) - }, - None => { - log_error!( - logger, - "Melt succeeded but no payment hash or preimage for quote: {quote_id}" - ); - PaymentHash([0u8; 32]) // Placeholder, should not happen - }, - } - }, - }; - - let payment_preimage = preimage.unwrap_or(PaymentPreimage([0u8; 32])); - - if tx_metadata - .set_preimage(payment_id, payment_preimage.0) - .await - .is_err() - { - log_error!( - logger, - "Failed to set preimage for payment {payment_id:?}" - ); - } - - let _ = event_queue - .add_event(Event::PaymentSuccessful { - payment_id, - payment_hash: hash, - payment_preimage, - fee_paid_msat, - }) - .await; - }, - MeltQuoteState::Failed => { - log_error!(logger, "Melt failed for quote: {quote_id}"); - let payment_id = PaymentId::Trusted(payment_id); - let is_rebalance = { - let map = tx_metadata.read(); - map.get(&payment_id).is_some_and(|m| m.ty.is_rebalance()) - }; - - if !is_rebalance { - let _ = event_queue - .add_event(Event::PaymentFailed { - payment_id, - payment_hash, - reason: None, - }) - .await; - } - }, - state => { - log_error!( - logger, - "Melt in unknown state {state} for quote: {quote_id}" - ); - // todo should we watch for it to complete? - }, - } - }, - Err(e) => { - log_error!(logger, "Failed to melt quote {quote_id}: {e}"); - let payment_id = PaymentId::Trusted(payment_id); - let is_rebalance = { - let map = tx_metadata.read(); - map.get(&payment_id).is_some_and(|m| m.ty.is_rebalance()) - }; - - if !is_rebalance { - let _ = event_queue - .add_event(Event::PaymentFailed { - payment_id, - payment_hash, - reason: None, - }) - .await; - } - }, + let resolved = melt + .handle_melt_result("e_id, payment_id, payment_hash, melt_result, submitted) + .await; + melt.in_flight.lock().unwrap().remove(&payment_id); + drop(gate); + if !resolved { + melt.reconcile.notify_one(); } }); } @@ -947,6 +888,279 @@ impl Cashu { } } +/// State shared between melt tasks and the reconciliation task. +struct MeltContext { + payments: Arc, + /// Payment IDs with a melt running in this process. + in_flight: Mutex>, + /// Melt tasks hold this shared; reconciliation holds it exclusively so it never touches a + /// saga while the melt that owns it is running. + gate: RwLock<()>, + /// Wakes the reconciliation task after a melt ended without a definite outcome. + reconcile: Notify, + event_queue: Arc, + tx_metadata: TxMetadataStore, + unit: CurrencyUnit, + logger: Arc, +} + +impl MeltContext { + /// Records the result of a melt request. Returns whether the outcome is final. + async fn handle_melt_result( + &self, quote_id: &str, payment_id: [u8; 32], payment_hash: Option, + result: Result, submitted: bool, + ) -> bool { + match result { + Ok(res) => self.handle_melt_outcome(quote_id, payment_id, payment_hash, &res).await, + Err(e) => { + log_error!(self.logger, "Failed to melt quote {quote_id}: {e}"); + if matches!(e, cdk::Error::PendingQuote | cdk::Error::PaidQuote) + || (submitted && !melt_was_rejected(&e)) + { + // The mint may have paid despite this error. Keep history pending. + return false; + } + self.melt_failed(payment_id, payment_hash).await; + true + }, + } + } + + /// Records the mint's answer for a melt. Returns whether the outcome is final. + async fn handle_melt_outcome( + &self, quote_id: &str, payment_id: [u8; 32], payment_hash: Option, + res: &FinalizedMelt, + ) -> bool { + match res.state() { + MeltQuoteState::Paid => { + log_info!(self.logger, "Successfully sent for quote: {quote_id}"); + self.melt_succeeded(quote_id, payment_id, payment_hash, res).await; + true + }, + // The CDK reports a melt it compensated without ever executing as unpaid. + MeltQuoteState::Failed | MeltQuoteState::Unpaid => { + log_error!(self.logger, "Melt failed for quote: {quote_id}"); + self.melt_failed(payment_id, payment_hash).await; + true + }, + state => { + log_info!(self.logger, "Melt still {state} for quote: {quote_id}"); + false + }, + } + } + + async fn melt_succeeded( + &self, quote_id: &str, payment_id: [u8; 32], payment_hash: Option, + res: &FinalizedMelt, + ) { + let first_report = self.payments.mark_completed(payment_id).await.unwrap_or_else(|e| { + log_error!(self.logger, "Failed to save payment success: {e}"); + true + }); + let fee_paid_msat = + convert_amount(res.fee_paid(), &self.unit).ok().map(|fee| fee.milli_sats()); + // Wake registered rebalances even when their public payment event is suppressed or + // cannot be persisted. The watcher ignores a repeated report itself. + if let Some(hash) = payment_hash { + let receipt = ReceivedLightningPayment { id: payment_id, fee_paid_msat }; + self.event_queue.rebalance_watchers.sent(hash.0, Some(receipt)); + } + if !first_report { + log_debug!(self.logger, "Success of quote {quote_id} was already reported"); + return; + } + let payment_id = PaymentId::Trusted(payment_id); + let is_rebalance = { + let map = self.tx_metadata.read(); + map.get(&payment_id).is_some_and(|m| m.ty.is_rebalance()) + }; + if is_rebalance { + return; + } + + let preimage: Option = match res.payment_proof() { + Some(str) => match FromHex::from_hex(str) { + Ok(b) => Some(PaymentPreimage(b)), + Err(e) => { + log_error!( + self.logger, + "Failed to decode preimage ({:?}) for quote {quote_id}: {e}", + res.payment_proof() + ); + None + }, + }, + None => { + // Expected for same-mint payments: when the melt's bolt11 destination is a + // mint quote on this same mint, cdk-mintd settles internally. No Lightning + // payment occurs, so there is no preimage to return. The success path below + // already tolerates None (hash falls back to the invoice payment_hash). + log_info!( + self.logger, + "Melt for quote {quote_id} settled without a preimage (internal/same-mint settlement)" + ); + None + }, + }; + + let hash = match payment_hash { + Some(hash) => hash, + None => match preimage { + Some(pre) => { + let hash = Sha256::hash(&pre.0); + PaymentHash(hash.to_byte_array()) + }, + None => { + log_error!( + self.logger, + "Melt succeeded but no payment hash or preimage for quote: {quote_id}" + ); + PaymentHash([0u8; 32]) // Placeholder, should not happen + }, + }, + }; + + let payment_preimage = preimage.unwrap_or(PaymentPreimage([0u8; 32])); + + if self.tx_metadata.set_preimage(payment_id, payment_preimage.0).await.is_err() { + log_error!(self.logger, "Failed to set preimage for payment {payment_id:?}"); + } + + let _ = self + .event_queue + .add_event(Event::PaymentSuccessful { + payment_id, + payment_hash: hash, + payment_preimage, + fee_paid_msat, + }) + .await; + } + + async fn melt_failed(&self, payment_id: [u8; 32], payment_hash: Option) { + let first_report = self.payments.mark_failed(payment_id).await.unwrap_or_else(|e| { + log_error!(self.logger, "Failed to save payment failure: {e}"); + true + }); + if let Some(hash) = payment_hash { + self.event_queue.rebalance_watchers.sent(hash.0, None); + } + if !first_report { + log_debug!(self.logger, "Failure of payment {payment_id:?} was already reported"); + return; + } + let payment_id = PaymentId::Trusted(payment_id); + let is_rebalance = { + let map = self.tx_metadata.read(); + map.get(&payment_id).is_some_and(|m| m.ty.is_rebalance()) + }; + if !is_rebalance { + let _ = self + .event_queue + .add_event(Event::PaymentFailed { payment_id, payment_hash, reason: None }) + .await; + } + } + + /// Resolves melts with an unknown outcome. Returns how many remain unresolved. + /// + /// Interrupted melts are finalized through the CDK's saga log, which asks the mint and + /// either recovers the change or releases the reserved proofs. Submitted payments the CDK + /// has no saga for are checked against the mint directly. Melts running in this process + /// are left alone so the two paths cannot race on the same quote. + async fn reconcile(&self, wallet: &Wallet) -> usize { + let mut unresolved = 0; + let Ok(_gate) = self.gate.try_write() else { + return 1; + }; + match wallet.finalize_pending_melts().await { + Ok(finalized) => { + for melt in finalized { + let quote_id = melt.quote_id().to_owned(); + let payment_id = Cashu::id_to_32_byte_array("e_id); + let payment_hash = quote_payment_hash(wallet, "e_id).await; + if !self.handle_melt_outcome("e_id, payment_id, payment_hash, &melt).await { + unresolved += 1; + } + } + }, + Err(e) => { + log_error!(self.logger, "Failed to finalize pending melts: {e}"); + unresolved += 1; + }, + } + + let pending = self.payments.pending().await; + if pending.is_empty() { + return unresolved; + } + // Quotes with an incomplete saga belong to the CDK; the pass above settles those. + let saga_quotes: HashSet = match wallet.localstore.get_incomplete_sagas().await { + Ok(sagas) => sagas.into_iter().filter_map(|saga| saga.quote_id).collect(), + Err(e) => { + log_error!(self.logger, "Failed to load incomplete sagas: {e}"); + return unresolved + pending.len(); + }, + }; + for (payment_id, reference) in pending { + if self.in_flight.lock().unwrap().contains(&payment_id) { + continue; + } + let Some(quote_id) = reference else { + log_warn!(self.logger, "No melt quote for pending payment {payment_id:?}"); + unresolved += 1; + continue; + }; + if saga_quotes.contains("e_id) { + unresolved += 1; + continue; + } + let quote = match wallet.localstore.get_melt_quote("e_id).await { + Ok(Some(quote)) => quote, + Ok(None) => { + log_warn!(self.logger, "Melt quote {quote_id} is missing"); + unresolved += 1; + continue; + }, + Err(e) => { + log_warn!(self.logger, "Failed to load melt quote {quote_id}: {e}"); + unresolved += 1; + continue; + }, + }; + let payment_hash = + Bolt11Invoice::from_str("e.request).ok().map(|i| i.payment_hash()); + let status = match wallet.check_melt_quote_status("e.id).await { + Ok(status) => status, + Err(e) => { + log_warn!(self.logger, "Failed to check melt quote {}: {e}", quote.id); + unresolved += 1; + continue; + }, + }; + let outcome = FinalizedMelt::new( + quote.id.clone(), + status.state, + status.payment_preimage.clone(), + status.amount, + CdkAmount::ZERO, + None, + ); + if !self.handle_melt_outcome("e.id, payment_id, payment_hash, &outcome).await { + unresolved += 1; + } + } + unresolved + } +} + +/// The payment hash of the invoice a melt quote pays, when it is a BOLT 11 invoice. +async fn quote_payment_hash(wallet: &Wallet, quote_id: &str) -> Option { + let quote = wallet.localstore.get_melt_quote(quote_id).await.ok().flatten()?; + Bolt11Invoice::from_str("e.request).ok().map(|i| i.payment_hash()) +} + fn convert_amount(cdk_amount: CdkAmount, unit: &CurrencyUnit) -> Result { match unit { CurrencyUnit::Sat => { @@ -961,6 +1175,35 @@ fn convert_amount(cdk_amount: CdkAmount, unit: &CurrencyUnit) -> Result bool { + matches!( + error, + cdk::Error::PaymentFailed + | cdk::Error::InsufficientFunds + | cdk::Error::InvalidInvoice + | cdk::Error::ExpiredQuote(_, _) + | cdk::Error::AmountOutofLimitRange(_, _, _) + | cdk::Error::MeltingDisabled + | cdk::Error::UnsupportedUnit + | cdk::Error::MaxFeeExceeded + ) +} + +#[cfg(test)] +mod melt_error_tests { + use super::*; + + #[test] + fn ambiguous_melt_errors_are_not_failures() { + assert!(melt_was_rejected(&cdk::Error::PaymentFailed)); + assert!(melt_was_rejected(&cdk::Error::InsufficientFunds)); + assert!(!melt_was_rejected(&cdk::Error::Timeout)); + assert!(!melt_was_rejected(&cdk::Error::PendingQuote)); + assert!(!melt_was_rejected(&cdk::Error::PaidQuote)); + assert!(!melt_was_rejected(&cdk::Error::Internal)); + } +} + #[cfg(test)] mod tests { use super::*; @@ -991,6 +1234,16 @@ mod tests { let db = Arc::new(CashuKvDatabase::new(store.shared(), Arc::clone(&runtime)).await.unwrap()); let wallet = Cashu { + melt: Arc::new(MeltContext { + payments: Arc::new(PaymentStore::new(store.shared(), test_logger()).await.unwrap()), + in_flight: Mutex::new(HashSet::new()), + gate: RwLock::new(()), + reconcile: Notify::new(), + event_queue: Arc::clone(&event_queue), + tx_metadata, + unit: CurrencyUnit::Sat, + logger: test_logger(), + }), cashu_wallet: Arc::new( Wallet::new("http://127.0.0.1:1", CurrencyUnit::Sat, db, [1; 64], None).unwrap(), ), @@ -1001,8 +1254,6 @@ mod tests { supports_bolt12: Arc::new(AtomicBool::new(false)), supports_mpp: Arc::new(AtomicBool::new(false)), mint_quote_sender: mpsc::channel(1).0, - event_queue: Arc::clone(&event_queue), - tx_metadata, runtime: Arc::clone(&runtime), npubcash_url: None, npub: None, diff --git a/orange-sdk/src/trusted_wallet/mod.rs b/orange-sdk/src/trusted_wallet/mod.rs index 9ce7834..68caa14 100644 --- a/orange-sdk/src/trusted_wallet/mod.rs +++ b/orange-sdk/src/trusted_wallet/mod.rs @@ -19,6 +19,9 @@ pub mod dummy; #[cfg(feature = "spark")] pub mod spark; +#[cfg(any(feature = "cashu", feature = "spark"))] +mod payment_store; + /// Represents a payment with its associated details. /// /// This struct contains information about a payment, including its unique ID, @@ -73,7 +76,10 @@ pub trait TrustedWalletInterface: Send + Sync + private::Sealed { /// the result of the payment, it should only initiate it and return the payment ID. /// /// This should later emit a [`PaymentSuccessful`] or [`PaymentFailed`] event - /// when the payment is completed or failed. + /// when the payment is completed or failed. A backend that is already paying the same + /// request may return that payment's ID instead of starting another; the terminal event is + /// still emitted once for that ID. When the backend cannot tell whether the payment went + /// through, the event follows once it has resolved the outcome. /// /// [`PaymentSuccessful`]: `crate::event::Event::PaymentSuccessful` /// [`PaymentFailed`]: `crate::event::Event::PaymentFailed` diff --git a/orange-sdk/src/trusted_wallet/payment_store.rs b/orange-sdk/src/trusted_wallet/payment_store.rs new file mode 100644 index 0000000..e69eabb --- /dev/null +++ b/orange-sdk/src/trusted_wallet/payment_store.rs @@ -0,0 +1,301 @@ +//! Keeps submitted payments visible when a trusted backend has no transaction for them. + +use super::{Payment, TrustedError}; +use crate::dyn_store::{DynStore, read_keys_bounded}; +use crate::logging::Logger; +use crate::store::{PaymentId, TxStatus}; +use bitcoin_payment_instructions::amount::Amount; +use ldk_node::lightning::impl_writeable_tlv_based; +use ldk_node::lightning::util::logger::Logger as _; +use ldk_node::lightning::util::persist::KVStore; +use ldk_node::lightning::util::ser::{Readable, Writeable}; +use ldk_node::lightning::{log_error, log_warn}; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::sync::Mutex; + +const PRIMARY: &str = "orange_sdk"; +const SECONDARY: &str = "trusted_payments"; + +#[derive(Clone)] +struct StoredPayment { + id: [u8; 32], + amount_msat: u64, + status: TxStatus, + time: u64, + /// The backend's own identifier for the payment, when it differs from `id`. + reference: Option, +} + +impl_writeable_tlv_based!(StoredPayment, { + (0, id, required), + (2, amount_msat, required), + (4, status, required), + (6, time, required), + (7, reference, option) +}); + +impl StoredPayment { + fn to_payment(&self) -> Payment { + Payment { + id: self.id, + amount: Amount::from_milli_sats(self.amount_msat).expect("Stored valid amount"), + fee: Amount::ZERO, + status: self.status, + outbound: true, + time_since_epoch: Duration::from_secs(self.time), + } + } +} + +pub(super) struct PaymentStore { + store: Arc, + logger: Arc, + payments: Mutex>, +} + +impl PaymentStore { + pub async fn new(store: Arc, logger: Arc) -> Result { + let keys = KVStore::list(store.as_ref(), PRIMARY, SECONDARY).await?; + let records = read_keys_bounded(Arc::clone(&store), PRIMARY, SECONDARY, keys).await?; + let mut payments = HashMap::with_capacity(records.len()); + for (key, bytes) in records { + // History metadata is not worth refusing to start the wallet over. + match StoredPayment::read(&mut &bytes[..]) { + Ok(payment) => { + payments.insert(payment.id, payment); + }, + Err(e) => log_error!(logger, "Skipping invalid stored trusted payment {key}: {e}"), + } + } + Ok(Self { store, logger, payments: Mutex::new(payments) }) + } + + /// Save before background work; return false if this ID is already submitted. + /// `reference` is the backend's own identifier, kept so the payment can be looked up later. + pub async fn insert_pending( + &self, id: [u8; 32], amount: Amount, reference: Option, + ) -> Result { + let mut payments = self.payments.lock().await; + if payments.get(&id).is_some_and(|p| p.status != TxStatus::Failed) { + return Ok(false); + } + let payment = StoredPayment { + id, + amount_msat: amount.milli_sats(), + status: TxStatus::Pending, + time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(), + reference, + }; + self.persist(&payment).await?; + payments.insert(id, payment); + Ok(true) + } + + /// Only call for a confirmed failure. Transport errors leave the payment pending. + /// Returns false if this outcome was already recorded. + pub async fn mark_failed(&self, id: [u8; 32]) -> Result { + self.mark(id, TxStatus::Failed).await + } + + /// Record a confirmed success until the backend lists the payment itself. + /// Returns false if this outcome was already recorded. + pub async fn mark_completed(&self, id: [u8; 32]) -> Result { + self.mark(id, TxStatus::Completed).await + } + + async fn mark(&self, id: [u8; 32], status: TxStatus) -> Result { + let mut payments = self.payments.lock().await; + let Some(mut payment) = payments.get(&id).cloned() else { + // Nothing recorded, so nothing has been reported for it either. + return Ok(true); + }; + if payment.status == status { + return Ok(false); + } + payment.status = status; + self.persist(&payment).await?; + payments.insert(id, payment); + Ok(true) + } + + /// Submitted payments whose outcome is not yet known, with their backend references. + pub async fn pending(&self) -> Vec<([u8; 32], Option)> { + self.payments + .lock() + .await + .values() + .filter(|p| p.status == TxStatus::Pending) + .map(|p| (p.id, p.reference.clone())) + .collect() + } + + async fn persist(&self, payment: &StoredPayment) -> Result<(), TrustedError> { + KVStore::write( + self.store.as_ref(), + PRIMARY, + SECONDARY, + &PaymentId::Trusted(payment.id).to_string(), + payment.encode(), + ) + .await?; + Ok(()) + } + + /// Backend terminal records take precedence and supply settled amounts and fees. Local + /// records they supersede are dropped so the store does not grow with every send. + /// + /// This runs on every history listing and rebalance check, so the backend list passes + /// through untouched when there is nothing local to merge. + pub async fn merge(&self, mut backend: Vec) -> Vec { + let mut local = self.payments.lock().await; + if local.is_empty() { + return backend; + } + let mut covered = HashSet::with_capacity(local.len()); + let mut pruned = Vec::new(); + backend.retain(|payment| { + let Some(record) = local.get(&payment.id) else { return true }; + if payment.status == TxStatus::Pending { + // A confirmed local failure outranks a backend record that never settled. + if record.status == TxStatus::Failed { + return false; + } + } else { + pruned.push(payment.id); + } + covered.insert(payment.id); + true + }); + backend.extend(local.values().filter(|p| !covered.contains(&p.id)).map(|p| p.to_payment())); + for id in pruned { + local.remove(&id); + let key = PaymentId::Trusted(id).to_string(); + if let Err(e) = + KVStore::remove(self.store.as_ref(), PRIMARY, SECONDARY, &key, false).await + { + log_warn!(self.logger, "Failed to prune trusted payment record {key}: {e}"); + } + } + backend + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::logging::LoggerType; + use ldk_node::io::sqlite_store::SqliteStore; + + fn store() -> Arc { + let path = std::env::temp_dir().join(format!( + "orange-trusted-payments-{}", + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos() + )); + Arc::new(SqliteStore::new(path, Some("payments.sqlite".to_owned()), None).unwrap()) + } + + fn logger() -> Arc { + Arc::new(Logger::new(&LoggerType::LogFacade).expect("logger")) + } + + #[tokio::test] + async fn submitted_and_failed_payments_survive_restart() { + let store = store(); + let payments = PaymentStore::new(Arc::clone(&store), logger()).await.unwrap(); + let amount = Amount::from_sats(100).unwrap(); + payments.insert_pending([1; 32], amount, Some("quote-1".to_owned())).await.unwrap(); + payments.insert_pending([2; 32], amount, None).await.unwrap(); + assert!(payments.mark_failed([1; 32]).await.unwrap()); + assert!(!payments.mark_failed([1; 32]).await.unwrap()); + drop(payments); + + let payments = PaymentStore::new(store, logger()).await.unwrap(); + let history = payments.merge(vec![]).await; + assert_eq!(history.len(), 2); + assert!(history.iter().all(|p| p.outbound && p.amount == amount)); + assert_eq!(history.iter().find(|p| p.id == [1; 32]).unwrap().status, TxStatus::Failed); + assert_eq!(history.iter().find(|p| p.id == [2; 32]).unwrap().status, TxStatus::Pending); + assert_eq!(payments.pending().await, vec![([2; 32], None)]); + assert!(!payments.insert_pending([2; 32], amount, None).await.unwrap()); + // A confirmed failure can be retried with the same backend ID. + payments.insert_pending([1; 32], amount, Some("quote-1".to_owned())).await.unwrap(); + assert_eq!(payments.pending().await.len(), 2); + assert!(payments.pending().await.contains(&([1; 32], Some("quote-1".to_owned())))); + assert_eq!(payments.merge(vec![]).await.len(), 2); + } + + #[tokio::test] + async fn backend_completion_replaces_local_record_without_duplicates() { + let store = store(); + let payments = PaymentStore::new(Arc::clone(&store), logger()).await.unwrap(); + let amount = Amount::from_sats(100).unwrap(); + payments.insert_pending([1; 32], amount, None).await.unwrap(); + payments.mark_failed([1; 32]).await.unwrap(); + let mut backend = Payment { + id: [1; 32], + amount, + fee: Amount::from_sats(2).unwrap(), + status: TxStatus::Pending, + outbound: true, + time_since_epoch: Duration::from_secs(1), + }; + assert_eq!(payments.merge(vec![backend.clone()]).await[0].status, TxStatus::Failed); + backend.status = TxStatus::Completed; + let history = payments.merge(vec![backend.clone()]).await; + assert_eq!(history.len(), 1); + assert_eq!(history[0].status, TxStatus::Completed); + assert_eq!(history[0].fee, backend.fee); + assert_eq!(history[0].time_since_epoch, backend.time_since_epoch); + + // The backend now owns the record, so the local copy is pruned, also on disk. + assert!(payments.merge(vec![]).await.is_empty()); + drop(payments); + let payments = PaymentStore::new(store, logger()).await.unwrap(); + assert!(payments.merge(vec![]).await.is_empty()); + } + + #[tokio::test] + async fn confirmed_success_is_listed_before_backend_catches_up() { + let payments = PaymentStore::new(store(), logger()).await.unwrap(); + let amount = Amount::from_sats(100).unwrap(); + payments.insert_pending([1; 32], amount, None).await.unwrap(); + assert!(payments.mark_completed([1; 32]).await.unwrap()); + let history = payments.merge(vec![]).await; + assert_eq!(history.len(), 1); + assert_eq!(history[0].status, TxStatus::Completed); + assert!(payments.pending().await.is_empty()); + // An outcome for a payment this store never saw still counts as new. + assert!(payments.mark_completed([9; 32]).await.unwrap()); + } + + #[tokio::test] + async fn merge_passes_backend_through_when_nothing_is_local() { + let payments = PaymentStore::new(store(), logger()).await.unwrap(); + let amount = Amount::from_sats(100).unwrap(); + let backend = vec![Payment { + id: [1; 32], + amount, + fee: Amount::ZERO, + status: TxStatus::Pending, + outbound: false, + time_since_epoch: Duration::from_secs(1), + }]; + let merged = payments.merge(backend.clone()).await; + assert_eq!(merged.len(), 1); + assert_eq!(merged[0].id, backend[0].id); + assert!(!merged[0].outbound); + } + + #[tokio::test] + async fn invalid_record_is_skipped_on_load() { + let store = store(); + KVStore::write(store.as_ref(), PRIMARY, SECONDARY, "garbage", vec![1, 2, 3]).await.unwrap(); + let payments = PaymentStore::new(Arc::clone(&store), logger()).await.unwrap(); + payments.insert_pending([1; 32], Amount::from_sats(1).unwrap(), None).await.unwrap(); + drop(payments); + let payments = PaymentStore::new(store, logger()).await.unwrap(); + assert_eq!(payments.merge(vec![]).await.len(), 1); + } +} diff --git a/orange-sdk/src/trusted_wallet/spark/mod.rs b/orange-sdk/src/trusted_wallet/spark/mod.rs index 2aefffb..e884ea5 100644 --- a/orange-sdk/src/trusted_wallet/spark/mod.rs +++ b/orange-sdk/src/trusted_wallet/spark/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod spark_store; +use super::payment_store::PaymentStore; use crate::bitcoin::Network; use crate::bitcoin::hex::FromHex; use crate::logging::Logger; @@ -95,6 +96,8 @@ impl SparkWalletConfig { /// A wallet implementation using the Breez Spark SDK. #[derive(Clone)] pub(crate) struct Spark { + tx_metadata: TxMetadataStore, + payments: Arc, event_queue: Arc, spark_wallet: Arc, shutdown_sender: watch::Sender<()>, @@ -161,7 +164,7 @@ impl TrustedWalletInterface for Spark { let payments = resp.payments.into_iter().map(|p| p.try_into()).collect::>()?; - Ok(payments) + Ok(self.payments.merge(payments).await) }) } @@ -220,6 +223,10 @@ impl TrustedWalletInterface for Spark { let prepare = self.spark_wallet.prepare_send_payment(params).await?; let uuid = Uuid::now_v7(); + let payment_id = parse_payment_id(&uuid.to_string())?; + self.payments.insert_pending(payment_id, amount, None).await?; + let payments = Arc::clone(&self.payments); + let tx_metadata = self.tx_metadata.clone(); // spawn payment send in background since it can take a while and we don't want to block the caller let w = Arc::clone(&self.spark_wallet); let logger = Arc::clone(&self.logger); @@ -243,11 +250,30 @@ impl TrustedWalletInterface for Spark { Err(e) => { log_error!(logger, "Failed to send payment: {e:?}"); event_queue.rebalance_watchers.sent(payment_hash, None); + if send_was_rejected(&e) { + if let Err(err) = payments.mark_failed(payment_id).await { + log_error!(logger, "Failed to save payment failure: {err}"); + } + let is_rebalance = tx_metadata + .read() + .get(&PaymentId::Trusted(payment_id)) + .is_some_and(|m| m.ty.is_rebalance()); + if is_rebalance { + return; + } + let _ = event_queue + .add_event(Event::PaymentFailed { + payment_id: PaymentId::Trusted(payment_id), + payment_hash: Some(PaymentHash(payment_hash)), + reason: None, + }) + .await; + } }, } }); - Ok(parse_payment_id(&uuid.to_string())?) + Ok(payment_id) } else { Err(TrustedError::UnsupportedOperation( "Only BOLT 11 is currently supported".to_owned(), @@ -326,6 +352,7 @@ impl Spark { }, }; + let payments = Arc::new(PaymentStore::new(Arc::clone(&store), Arc::clone(&logger)).await?); let spark_store = Arc::new(spark_store::SparkStore::new(store)); let builder = SdkBuilder::new(spark_config, seed).with_storage(spark_store); @@ -340,7 +367,7 @@ impl Spark { let listener = SparkEventHandler { event_queue: Arc::clone(&event_queue), - tx_metadata, + tx_metadata: tx_metadata.clone(), logger: Arc::clone(&logger), }; @@ -355,7 +382,15 @@ impl Spark { log_info!(logger, "Spark wallet initialized"); - Ok(Spark { spark_wallet, shutdown_sender, event_queue, runtime, logger }) + Ok(Spark { + tx_metadata, + payments, + spark_wallet, + shutdown_sender, + event_queue, + runtime, + logger, + }) } } @@ -637,6 +672,28 @@ impl TryFrom for Payment { } } +// A service or transport error does not prove that the payment failed. +fn send_was_rejected(error: &SdkError) -> bool { + matches!( + error, + SdkError::InvalidInput(_) | SdkError::InvalidUuid(_) | SdkError::InsufficientFunds + ) +} + +#[cfg(test)] +mod send_error_tests { + use super::*; + + #[test] + fn ambiguous_send_errors_are_not_failures() { + assert!(send_was_rejected(&SdkError::InsufficientFunds)); + assert!(send_was_rejected(&SdkError::InvalidInput("invalid".to_owned()))); + assert!(!send_was_rejected(&SdkError::NetworkError("timeout".to_owned()))); + assert!(!send_was_rejected(&SdkError::StorageError("write failed".to_owned()))); + assert!(!send_was_rejected(&SdkError::SparkError("unknown".to_owned()))); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/orange-sdk/tests/integration_tests.rs b/orange-sdk/tests/integration_tests.rs index e5ebf1d..dee77a2 100644 --- a/orange-sdk/tests/integration_tests.rs +++ b/orange-sdk/tests/integration_tests.rs @@ -6,10 +6,13 @@ use bitcoin_payment_instructions::http_resolver::HTTPHrnResolver; use bitcoin_payment_instructions::{ParseError, PaymentInstructions}; use ldk_node::NodeError; use ldk_node::bitcoin::Network; -use ldk_node::lightning_invoice::{Bolt11InvoiceDescription, Description}; +use ldk_node::bitcoin::secp256k1::{Secp256k1, SecretKey}; +use ldk_node::lightning_invoice::{ + Bolt11InvoiceDescription, Currency, Description, InvoiceBuilder, PaymentHash, PaymentSecret, +}; use ldk_node::payment::{ConfirmationStatus, PaymentDirection, PaymentStatus}; use log::info; -use orange_sdk::{Event, PaymentInfo, PaymentType, TxStatus, WalletError}; +use orange_sdk::{Event, PaymentId, PaymentInfo, PaymentType, TxStatus, WalletError}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -2017,6 +2020,205 @@ async fn test_invalid_payment_instructions() { .await; } +#[tokio::test(flavor = "multi_thread")] +#[test_log::test] +async fn test_failed_lightning_send_is_surfaced() { + test_utils::run_test(|params| async move { + let wallet = Arc::clone(¶ms.wallet); + let bitcoind = Arc::clone(¶ms.bitcoind); + let electrsd = Arc::clone(¶ms.electrsd); + let third_party = Arc::clone(¶ms.third_party); + + // Give the wallet a spendable lightning balance. + open_channel_from_lsp(&wallet, Arc::clone(&third_party)).await; + generate_blocks(&bitcoind, &electrsd, 6).await; + test_utils::wait_for_condition("wallet sync after channel open", || async { + wallet.channels().iter().any(|c| c.confirmations.is_some_and(|n| n > 0) && c.is_usable) + }) + .await; + + // An invoice signed by a key no node ever announced: route-finding fails + // synchronously, and ldk-node records the attempt as a failed outbound payment. + // Unlike the rejected payments above (which never reach the node and leave no + // record), this attempt must show up in the transaction list. + let pay_amt = Amount::from_sats(10_000).unwrap(); + let secp = Secp256k1::new(); + let no_such_node = SecretKey::from_slice(&[99; 32]).unwrap(); + let invoice = InvoiceBuilder::new(Currency::Regtest) + .description("no route to this payee".to_string()) + .payment_hash(PaymentHash([43; 32])) + .payment_secret(PaymentSecret([44; 32])) + .current_timestamp() + .min_final_cltv_expiry_delta(144) + .amount_milli_satoshis(pay_amt.milli_sats()) + .build_signed(|hash| secp.sign_ecdsa_recoverable(hash, &no_such_node)) + .unwrap(); + + let instr = wallet.parse_payment_instructions(invoice.to_string().as_str()).await.unwrap(); + let info = PaymentInfo::build(instr, None).unwrap(); + let res = wallet.pay(&info).await; + assert!( + matches!(res, Err(WalletError::LdkNodeFailure(NodeError::PaymentSendingFailed))), + "send to an unroutable payee must fail synchronously, got {res:?}" + ); + + let txs = wallet.list_transactions().await.unwrap(); + let outbound: Vec<_> = txs.iter().filter(|t| t.outbound).collect(); + assert_eq!(outbound.len(), 1, "the failed send should surface exactly once: {txs:?}"); + let failed = outbound[0]; + assert_eq!(failed.status, TxStatus::Failed); + assert_eq!(failed.amount, Some(pay_amt)); + match &failed.payment_type { + PaymentType::OutgoingLightningBolt11 { payment_preimage } => { + assert!(payment_preimage.is_none(), "a failed payment has no preimage"); + }, + pt => panic!("Payment type should be OutgoingLightningBolt11, got {pt:?}"), + } + }) + .await; +} + +#[cfg(feature = "_cashu-tests")] +#[tokio::test(flavor = "multi_thread")] +#[test_log::test] +async fn test_failed_cashu_send_is_surfaced() { + test_utils::run_test(|params| async move { + let wallet = ¶ms.wallet; + let funded = Amount::from_sats(500).unwrap(); + let uri = wallet.get_single_use_receive_uri(Some(funded)).await.unwrap(); + assert!(uri.from_trusted); + params.third_party.bolt11_payment().send(&uri.invoice, None).unwrap(); + test_utils::wait_for_condition("trusted wallet funded", || async { + wallet.get_balance().await.unwrap().trusted == funded + }) + .await; + assert!(matches!(wait_next_event(wallet).await, Event::PaymentReceived { .. })); + + let amount = Amount::from_sats(100).unwrap(); + let secp = Secp256k1::new(); + let key = SecretKey::from_slice(&[99; 32]).unwrap(); + let invoice = InvoiceBuilder::new(Currency::Regtest) + .description("unreachable payee".to_owned()) + .payment_hash(PaymentHash([43; 32])) + .payment_secret(PaymentSecret([44; 32])) + .current_timestamp() + .min_final_cltv_expiry_delta(144) + .amount_milli_satoshis(amount.milli_sats()) + .build_signed(|hash| secp.sign_ecdsa_recoverable(hash, &key)) + .unwrap(); + let info = PaymentInfo::build( + wallet.parse_payment_instructions(&invoice.to_string()).await.unwrap(), + None, + ) + .unwrap(); + let id = wallet.pay(&info).await.unwrap(); + assert!(matches!(id, PaymentId::Trusted(_))); + let history = wallet.list_transactions().await.unwrap(); + assert_eq!(history.iter().filter(|t| t.id == id).count(), 1); + assert!(matches!(wait_next_event(wallet).await, + Event::PaymentFailed { payment_id, .. } if payment_id == id)); + let history = wallet.list_transactions().await.unwrap(); + let outbound: Vec<_> = history.iter().filter(|t| t.outbound).collect(); + assert_eq!(outbound.len(), 1, "{history:?}"); + assert_eq!(outbound[0].id, id); + assert_eq!(outbound[0].status, TxStatus::Failed); + assert_eq!(outbound[0].amount, Some(amount)); + }) + .await; +} + +#[tokio::test(flavor = "multi_thread")] +#[test_log::test] +#[cfg_attr( + feature = "_cashu-tests", + ignore = "CDK's test mint/payment processor does not support partial MPP melts" +)] +async fn test_failed_mpp_lightning_leg_is_not_listed_separately() { + test_utils::run_test(|params| async move { + let wallet = Arc::clone(¶ms.wallet); + let bitcoind = Arc::clone(¶ms.bitcoind); + let third_party = Arc::clone(¶ms.third_party); + let electrsd = Arc::clone(¶ms.electrsd); + let lsp = Arc::clone(¶ms.lsp); + let desc = Bolt11InvoiceDescription::Direct(Description::empty()); + + // Fund the trusted wallet with 100 sats before a channel exists (once inbound + // liquidity exists, small receives route to the lightning wallet instead). + let trusted_amt = Amount::from_sats(100).unwrap(); + let uri = wallet.get_single_use_receive_uri(Some(trusted_amt)).await.unwrap(); + assert!(uri.from_trusted); + third_party.bolt11_payment().send(&uri.invoice, None).unwrap(); + test_utils::wait_for_condition("trusted balance funded", || async { + wallet.get_balance().await.unwrap().trusted == trusted_amt + }) + .await; + assert!(matches!(wait_next_event(&wallet).await, Event::PaymentReceived { .. })); + + // Open a lightning channel. + open_channel_from_lsp(&wallet, Arc::clone(&third_party)).await; + generate_blocks(&bitcoind, &electrsd, 6).await; + test_utils::wait_for_condition("wallet sync after channel open", || async { + wallet.channels().iter().any(|c| c.confirmations.is_some_and(|n| n > 0) && c.is_usable) + }) + .await; + + // Drain spendable lightning liquidity down to ~150 sats. The channel reserve + // keeps the *total* lightning balance well above that, which is exactly the + // gap this test needs: the MPP split passes the balance check, but the + // lightning leg exceeds what a route can actually carry and fails + // synchronously with RouteNotFound after the trusted leg is already in + // flight. + let sendable = + wallet.channels().iter().find(|c| c.is_usable).unwrap().next_outbound_htlc_limit_msat; + let drain = lsp.bolt11_payment().receive(sendable - 150_000, &desc, 300).unwrap(); + let drain_info = PaymentInfo::build( + wallet.parse_payment_instructions(&drain.to_string()).await.unwrap(), + None, + ) + .unwrap(); + wallet.pay(&drain_info).await.unwrap(); + assert!(matches!(wait_next_event(&wallet).await, Event::PaymentSuccessful { .. })); + test_utils::wait_for_condition("lightning balance drained below 200 sats", || async { + wallet + .channels() + .iter() + .find(|c| c.is_usable) + .is_some_and(|c| c.next_outbound_htlc_limit_msat < 200_000) + }) + .await; + + // 350 sats = 100 trusted + 250 lightning. The 250 sat lightning leg exceeds + // the ~150 sats of usable outbound liquidity, so it fails synchronously. + let pay_amt = Amount::from_sats(350).unwrap(); + let invoice = + third_party.bolt11_payment().receive(pay_amt.milli_sats(), &desc, 300).unwrap(); + let info = PaymentInfo::build( + wallet.parse_payment_instructions(&invoice.to_string()).await.unwrap(), + Some(pay_amt), + ) + .unwrap(); + assert!(wallet.pay(&info).await.is_err(), "MPP with an unroutable lightning leg must fail"); + + // The failed lightning leg is internal bookkeeping of the MPP attempt, which + // is surfaced through the trusted leg. It must not appear as an additional, + // standalone failed payment (the only other outbound row is the drain above). + let txs = wallet.list_transactions().await.unwrap(); + let failed_outbound = + txs.iter().filter(|t| t.outbound && t.status == TxStatus::Failed).count(); + assert_eq!( + failed_outbound, 0, + "the failed MPP lightning leg must not list on its own, got {txs:?}" + ); + let trusted_legs = + txs.iter().filter(|t| t.outbound && matches!(t.id, PaymentId::Trusted(_))).count(); + assert_eq!( + trusted_legs, 1, + "the MPP attempt should surface via the trusted leg, got {txs:?}" + ); + }) + .await; +} + #[tokio::test(flavor = "multi_thread")] #[test_log::test] async fn test_payment_with_expired_invoice() { @@ -2805,3 +3007,63 @@ async fn test_lsp_connectivity_fallback() { }) .await; } + +#[tokio::test(flavor = "multi_thread")] +#[test_log::test] +#[cfg_attr( + feature = "_cashu-tests", + ignore = "CDK's test mint/payment processor does not support partial MPP melts" +)] +async fn test_duplicate_mpp_preserves_completed_payment() { + test_utils::run_test(|params| async move { + let wallet = Arc::clone(¶ms.wallet); + let bitcoind = Arc::clone(¶ms.bitcoind); + let third_party = Arc::clone(¶ms.third_party); + let electrsd = Arc::clone(¶ms.electrsd); + let desc = Bolt11InvoiceDescription::Direct(Description::empty()); + + // Fund the trusted wallet with 100 sats before a channel exists (once inbound + // liquidity exists, small receives route to the lightning wallet instead). + let trusted_amt = Amount::from_sats(100).unwrap(); + let uri = wallet.get_single_use_receive_uri(Some(trusted_amt)).await.unwrap(); + assert!(uri.from_trusted); + third_party.bolt11_payment().send(&uri.invoice, None).unwrap(); + test_utils::wait_for_condition("trusted balance funded", || async { + wallet.get_balance().await.unwrap().trusted == trusted_amt + }) + .await; + assert!(matches!(wait_next_event(&wallet).await, Event::PaymentReceived { .. })); + + // Open a lightning channel. + open_channel_from_lsp(&wallet, Arc::clone(&third_party)).await; + generate_blocks(&bitcoind, &electrsd, 6).await; + test_utils::wait_for_condition("wallet sync after channel open", || async { + wallet.channels().iter().any(|c| c.confirmations.is_some_and(|n| n > 0) && c.is_usable) + }) + .await; + + let pay_amt = Amount::from_sats(350).unwrap(); + let invoice = + third_party.bolt11_payment().receive(pay_amt.milli_sats(), &desc, 300).unwrap(); + let info = PaymentInfo::build( + wallet.parse_payment_instructions(&invoice.to_string()).await.unwrap(), + Some(pay_amt), + ) + .unwrap(); + let original_id = wallet.pay(&info).await.unwrap(); + assert!(matches!(wait_next_event(&wallet).await, Event::PaymentSuccessful { .. })); + let before = wallet.list_transactions().await.unwrap(); + assert!(before.iter().any(|t| t.id == original_id && t.status == TxStatus::Completed)); + let retry = wallet.pay(&info).await; + assert!( + matches!(retry, Err(WalletError::LdkNodeFailure(NodeError::DuplicatePayment))), + "{retry:?}" + ); + let after = wallet.list_transactions().await.unwrap(); + assert!( + after.iter().any(|t| t.id == original_id && t.status == TxStatus::Completed), + "Retry removed completed payment {original_id:?}: {after:?}" + ); + }) + .await; +}