From e31fc19dac059b70bad38af888c5b8b40f5f50ae Mon Sep 17 00:00:00 2001 From: Hash Money Date: Thu, 10 Sep 2026 19:54:07 -0700 Subject: [PATCH 1/2] Surface failed and pending outbound Lightning payments in list_transactions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ldk-node records a synchronously-failed send as a Failed outbound PaymentDetails before returning the error (bolt11 send_internal's SendingFailed arm; bolt12 and spontaneous likewise), and pay() only writes tx metadata when the send returns Ok — so a failed send lands in the no-metadata branch of list_transactions and was hidden by the status != Completed filter, leaving no trace in the transaction history. Surface outbound attempts regardless of status; non-completed inbound records (issued-but-unpaid invoices) stay hidden. The no-metadata branch's debug assertion assumed outbound records always carry metadata; that only holds for successful sends, so it fires today in any debug build that lists transactions after a synchronously-failed send. Replace it with the trusted loop's pattern (log_warn plus a _test-utils-gated assert), scoped to Succeeded — and even Succeeded can legitimately lack metadata after a crash between ldk-node's persist and the metadata write, hence warn-and-assert. Surfacing outbound records also exposed one internal leg: when try_mpp_bolt11's lightning portion fails synchronously after the trusted leg is in flight, ldk-node has recorded a failed outbound payment for it that would list as a second, standalone transaction. Nothing is in flight after a synchronous failure, so the MPP error path now removes that record; the attempt stays surfaced through the trusted leg, as that error path already intends. Trusted-backend records are unchanged: failed Spark sends either already surface through the metadata branch (keyed by the idempotency uuid, which never filtered on status) or never reach storage, and the CDK only records melts that reached Paid. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016zApnExfnDXm1Kxu8pf9P9 --- orange-sdk/src/lib.rs | 113 +++++++++++++++--- orange-sdk/tests/integration_tests.rs | 157 +++++++++++++++++++++++++- 2 files changed, 254 insertions(+), 16 deletions(-) diff --git a/orange-sdk/src/lib.rs b/orange-sdk/src/lib.rs index 69a99ed..dd65f83 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; } @@ -1080,12 +1089,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) @@ -1582,6 +1607,24 @@ impl Wallet { Ok(id) => id, Err(e) => { log_error!(self.inner.logger, "Failed to send lightning MPP portion: {e:?}"); + // The lightning leg failed synchronously, so nothing is in flight — but + // ldk-node has still recorded a failed outbound payment for it, keyed by + // the invoice's payment hash. Remove that record: it is an internal MPP + // leg, not an independent payment, and the attempt is surfaced through + // the trusted leg below. + use ldk_node::lightning::ln::channelmanager::PaymentId as LdkPaymentId; + if let Err(remove_err) = self + .inner + .ln_wallet + .inner + .ldk_node + .remove_payment(&LdkPaymentId(payment_hash.0)) + { + log_error!( + self.inner.logger, + "Failed to remove failed MPP lightning leg record: {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 +1850,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 +1888,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/tests/integration_tests.rs b/orange-sdk/tests/integration_tests.rs index e5ebf1d..ab3f654 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,156 @@ 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; +} + +#[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() { From 266a770be33b7bc3e94273fca357b5f6e1e97721 Mon Sep 17 00:00:00 2001 From: benthecarman Date: Mon, 14 Sep 2026 17:23:49 -0500 Subject: [PATCH 2/2] Surface failed Spark and Cashu payments Keep submitted trusted payments visible while pending and after a confirmed failure, even when the backend has no transaction record. Merge local records with backend history using existing payment IDs, leaving uncertain send outcomes pending until they are resolved. Resolve trusted sends whose outcome was unknown instead of leaving them pending forever. Cashu finalizes interrupted melts through the CDK saga log at startup and whenever a melt ends without a definite result, and asks the mint about submitted payments the CDK has no saga for. Melts in this process are excluded so the two paths cannot race on the same quote. Every melt outcome, including a reconciled one, reports to the rebalance watcher, so a rebalance waits for the real result instead of being released on an ambiguous error. Melt an unpaid quote again after a transient error instead of returning the old payment ID without doing anything; the CDK and the mint reject a quote that is actually in flight. Reject a duplicate MPP invoice before the trusted leg goes out, since that leg cannot be recalled, and find the lightning MPP leg among recent outbound records by hash and direction rather than deriving an LDK payment ID from the hash. Only the newest page of the payment list is read: a synchronously failed leg was inserted moments earlier, and the duplicate check is best effort. Preserve pending and completed LDK records when an MPP leg is rejected as a duplicate payment. Keep the payment store cheap on the paths that run often. History listings and rebalance checks pass the backend list through untouched when nothing is recorded locally, records carry the backend's own quote ID so reconciliation looks a quote up directly instead of scanning them all, and an outcome is reported once even if two paths observe it. Prune local records once the backend lists the payment, skip corrupt records instead of failing wallet start, and document that a backend may return an in-flight payment's ID; recording metadata for such an ID must not assert that the entry is new. Errors before submission do not create history entries. Co-Authored-By: Claude Fable 5.1 --- orange-sdk/src/lib.rs | 59 +- orange-sdk/src/lightning_wallet.rs | 16 + orange-sdk/src/trusted_wallet/cashu/mod.rs | 575 +++++++++++++----- orange-sdk/src/trusted_wallet/mod.rs | 8 +- .../src/trusted_wallet/payment_store.rs | 301 +++++++++ orange-sdk/src/trusted_wallet/spark/mod.rs | 65 +- orange-sdk/tests/integration_tests.rs | 109 ++++ 7 files changed, 948 insertions(+), 185 deletions(-) create mode 100644 orange-sdk/src/trusted_wallet/payment_store.rs diff --git a/orange-sdk/src/lib.rs b/orange-sdk/src/lib.rs index dd65f83..08b2530 100644 --- a/orange-sdk/src/lib.rs +++ b/orange-sdk/src/lib.rs @@ -839,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(), @@ -1340,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() }, @@ -1581,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 @@ -1607,23 +1626,27 @@ impl Wallet { Ok(id) => id, Err(e) => { log_error!(self.inner.logger, "Failed to send lightning MPP portion: {e:?}"); - // The lightning leg failed synchronously, so nothing is in flight — but - // ldk-node has still recorded a failed outbound payment for it, keyed by - // the invoice's payment hash. Remove that record: it is an internal MPP - // leg, not an independent payment, and the attempt is surfaced through - // the trusted leg below. - use ldk_node::lightning::ln::channelmanager::PaymentId as LdkPaymentId; - if let Err(remove_err) = self - .inner - .ln_wallet - .inner - .ldk_node - .remove_payment(&LdkPaymentId(payment_hash.0)) - { - log_error!( - self.inner.logger, - "Failed to remove failed MPP lightning leg record: {remove_err:?}" - ); + // 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 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 ab3f654..dee77a2 100644 --- a/orange-sdk/tests/integration_tests.rs +++ b/orange-sdk/tests/integration_tests.rs @@ -2078,6 +2078,55 @@ async fn test_failed_lightning_send_is_surfaced() { .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( @@ -2958,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; +}