diff --git a/src/builder.rs b/src/builder.rs index dc41aef1a..5a2b89bc0 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -2375,6 +2375,7 @@ fn build_with_store_internal( scorer, peer_store, payment_store, + pending_payment_store, lnurl_auth, is_running, node_metrics, diff --git a/src/channel/mod.rs b/src/channel/mod.rs new file mode 100644 index 000000000..3214b5253 --- /dev/null +++ b/src/channel/mod.rs @@ -0,0 +1,557 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Retrying user-initiated splices that LDK dropped before durably recording them. + +use std::ops::Deref; +use std::sync::Arc; + +use bitcoin::secp256k1::PublicKey; +use bitcoin::{Amount, OutPoint}; +use lightning::events::NegotiationFailureReason; +use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; +use lightning::ln::types::ChannelId; + +use crate::data_store::StorableObject; +use crate::event::{Event, EventQueue}; +use crate::fee_estimator::{ + max_funding_feerate, ConfirmationTarget, FeeEstimator, OnchainFeeEstimator, +}; +use crate::logger::{log_error, log_info, LdkLogger}; +use crate::payment::pending_payment_store::{ + PendingPaymentDetails, PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, + MAX_SPLICE_ATTEMPTS, +}; +use crate::payment::store::PaymentDetails; +use crate::payment::PaymentStatus; +use crate::types::{ChannelManager, PaymentStore, PendingPaymentStore, UserChannelId, Wallet}; +use crate::Error; + +/// The action to take on a `SpliceNegotiationFailed` for a splice intent we track, decided purely +/// from the failure `reason` and the intent's attempt count so the decision matrix can be +/// unit-tested without a live channel. A failure for a splice we don't track is surfaced directly +/// (see [`SpliceRetrier::on_negotiation_failed`]) and never reaches here. +#[derive(Debug, PartialEq, Eq)] +enum RetryDecision { + /// Give up: clear the intent and surface the failure to the user. + Abandon, + /// Resubmit the stored contribution unchanged (a transient failure such as a disconnect). + ResubmitStored, + /// Rebuild a fresh contribution from the original parameters (the stored one went stale). + Rebuild, +} + +fn decide_retry(reason: &NegotiationFailureReason, attempts: u8) -> RetryDecision { + if !reason.is_retriable() || attempts >= MAX_SPLICE_ATTEMPTS { + return RetryDecision::Abandon; + } + match reason { + // The stored contribution is still valid after a transient failure. + NegotiationFailureReason::PeerDisconnected | NegotiationFailureReason::Unknown => { + RetryDecision::ResubmitStored + }, + // The remaining retriable reasons (`FeeRateTooLow`, `ContributionInvalid`) mean the stored + // contribution went stale. + _ => RetryDecision::Rebuild, + } +} + +/// Resubmits user-initiated splices that LDK dropped before durably recording them. +/// +/// LDK only persists a splice once its negotiation reaches `AwaitingSignatures`, and it abandons an +/// earlier negotiation whenever the peer disconnects (which includes restarting the node). The +/// splice entry points persist a [`SpliceIntent`] before handing the contribution to LDK; this type +/// drives that intent back into [`ChannelManager::funding_contributed`] until the splice either +/// locks (clearing the intent) or fails for a reason retrying cannot address. +/// +/// Resubmitting does not require the peer to be connected: LDK holds on to the contribution and +/// initiates quiescence once the peer reconnects. +/// +/// [`ChannelManager::funding_contributed`]: lightning::ln::channelmanager::ChannelManager::funding_contributed +pub(crate) struct SpliceRetrier +where + L::Target: LdkLogger, +{ + channel_manager: Arc, + wallet: Arc, + fee_estimator: Arc, + pending_payment_store: Arc, + payment_store: Arc, + event_queue: Arc>, + logger: L, +} + +impl SpliceRetrier +where + L::Target: LdkLogger, +{ + pub(crate) fn new( + channel_manager: Arc, wallet: Arc, + fee_estimator: Arc, pending_payment_store: Arc, + payment_store: Arc, event_queue: Arc>, logger: L, + ) -> Self { + Self { + channel_manager, + wallet, + fee_estimator, + pending_payment_store, + payment_store, + event_queue, + logger, + } + } + + /// Reconciles persisted splice intents against live channel state. Run once at startup to pick + /// up splices LDK dropped before durably recording them — including those lost to a crash before + /// LDK persisted anything. + pub(crate) async fn reconcile(&self) { + let records = self.pending_payment_store.list_filter(|p| p.splice_intent().is_some()); + for record in records { + let id = record.id(); + let has_payment = self.has_payment(&record); + let Some(intent) = record.splice_intent().cloned() else { + continue; + }; + + let channel = self + .channel_manager + .list_channels_with_counterparty(&intent.counterparty_node_id) + .into_iter() + .find(|c| c.user_channel_id == intent.user_channel_id.0); + let channel = match channel { + Some(channel) => channel, + None => { + // The channel is gone; there is nothing to splice anymore. + self.clear_intent(id, has_payment).await; + continue; + }, + }; + + if channel.funding_txo != Some(intent.pre_splice_funding_txo) { + // The funding moved on, so the splice (or a replacement) locked. + self.clear_intent(id, has_payment).await; + continue; + } + + // `splice_channel` is a read-only probe of LDK's splice state. It fails when we already + // have a splice in flight (a held contribution, an in-progress negotiation, or one + // awaiting signatures), all of which LDK drives to completion on its own. + let template = match self + .channel_manager + .splice_channel(&channel.channel_id, &intent.counterparty_node_id) + { + Ok(template) => template, + Err(_) => continue, + }; + + // LDK persists a splice once negotiated, so a prior contribution means the intent was + // carried out — unless the intent was a fee bump at a higher feerate than negotiated. + let should_retry = match (&intent.kind, template.prior_contribution()) { + (SpliceKind::Rbf {}, Some(prior)) => { + prior.feerate() < intent.contribution.feerate() + }, + (SpliceKind::Rbf {}, None) => { + // The splice to bump is gone entirely; surface rather than guess. + self.abandon(id, has_payment, &intent).await; + continue; + }, + (_, Some(_)) => false, + (_, None) => true, + }; + if !should_retry { + continue; + } + + if intent.attempts >= MAX_SPLICE_ATTEMPTS { + self.abandon(id, has_payment, &intent).await; + continue; + } + + log_info!( + self.logger, + "Resubmitting splice for channel {} with counterparty {}", + channel.channel_id, + intent.counterparty_node_id, + ); + let counterparty_node_id = intent.counterparty_node_id; + let _ = self.submit(id, &channel.channel_id, &counterparty_node_id, intent).await; + } + } + + /// Persists the incremented attempt count and hands the contribution back to LDK. The count is + /// persisted first so that a crash mid-submission cannot lead to unbounded retries. + async fn submit( + &self, id: PaymentId, channel_id: &ChannelId, counterparty_node_id: &PublicKey, + mut intent: SpliceIntent, + ) -> Result<(), Error> { + intent.attempts += 1; + let contribution = intent.contribution.clone(); + let update = PendingPaymentDetailsUpdate { + id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent)), + }; + self.pending_payment_store.update(update).await?; + + self.channel_manager + .funding_contributed(channel_id, counterparty_node_id, contribution, None) + .map_err(|e| { + log_error!( + self.logger, + "Failed to resubmit splice for channel {} with counterparty {}: {:?}", + channel_id, + counterparty_node_id, + e, + ); + Error::ChannelSplicingFailed + }) + } + + /// Drops a splice intent: removes a record with no classified funding payment behind it + /// entirely, or keeps the record — with the intent cleared — when a payment exists, so the + /// payment keeps graduating. + async fn clear_intent(&self, id: PaymentId, has_payment: bool) { + if has_payment { + let _ = self + .pending_payment_store + .mutate(&id, |existing| { + record_with_intent_cleared(existing, self.payment_store.get(&id)) + }) + .await; + } else { + let _ = self.pending_payment_store.remove(&id).await; + } + } + + /// Gives up on a splice intent and surfaces the failure to the user. + async fn abandon(&self, id: PaymentId, has_payment: bool, intent: &SpliceIntent) { + log_error!( + self.logger, + "Abandoning splice for channel {} with counterparty {}", + intent.channel_id, + intent.counterparty_node_id, + ); + self.clear_intent(id, has_payment).await; + let event = Event::SpliceNegotiationFailed { + channel_id: intent.channel_id, + user_channel_id: intent.user_channel_id, + counterparty_node_id: intent.counterparty_node_id, + }; + if let Err(e) = self.event_queue.add_event(event).await { + log_error!(self.logger, "Failed to push to event queue: {}", e); + } + } + + /// Applies a `SpliceNegotiationFailed` to any matching splice intent, retrying recoverable + /// failures. Returns whether the failure should be surfaced to the user (i.e. the splice is + /// given up on). + pub(crate) async fn on_negotiation_failed( + &self, user_channel_id: UserChannelId, reason: NegotiationFailureReason, + contribution: Option, + ) -> bool { + let Some(record) = self.record_for_channel(user_channel_id) else { + return true; + }; + let id = record.id(); + let has_payment = self.has_payment(&record); + let Some(intent) = record.splice_intent().cloned() else { + return true; + }; + + // Only act on failures of the splice we are tracking. A mismatch means the failure concerns + // some other attempt (e.g. a stale event replayed after a newer splice was initiated). + if contribution.as_ref() != Some(&intent.contribution) { + return true; + } + + let channel_id = intent.channel_id; + let counterparty_node_id = intent.counterparty_node_id; + match decide_retry(&reason, intent.attempts) { + RetryDecision::Abandon => { + self.clear_intent(id, has_payment).await; + true + }, + RetryDecision::ResubmitStored => { + // The same contribution remains valid; resubmit it. Skip if LDK already has a splice + // in flight for this channel (e.g. the startup reconciler resubmitted first). + if self.channel_manager.splice_channel(&channel_id, &counterparty_node_id).is_err() + { + return false; + } + log_info!( + self.logger, + "Resubmitting splice for channel {} with counterparty {} after a recoverable failure", + channel_id, + counterparty_node_id, + ); + let _ = self.submit(id, &channel_id, &counterparty_node_id, intent).await; + false + }, + RetryDecision::Rebuild => { + // The stored contribution went stale; rebuild a fresh one from the original params. + match self + .rebuild_contribution(&channel_id, &counterparty_node_id, &intent.kind) + .await + { + Ok(contribution) => { + log_info!( + self.logger, + "Resubmitting rebuilt splice for channel {} with counterparty {}", + channel_id, + counterparty_node_id, + ); + let mut intent = intent; + intent.contribution = contribution; + let _ = self.submit(id, &channel_id, &counterparty_node_id, intent).await; + false + }, + Err(e) => { + log_error!( + self.logger, + "Abandoning splice for channel {}: failed to rebuild contribution: {:?}", + channel_id, + e, + ); + self.clear_intent(id, has_payment).await; + true + }, + } + }, + } + } + + /// Clears any splice intent made obsolete by a newly locked funding transaction. + pub(crate) async fn on_channel_ready( + &self, user_channel_id: UserChannelId, funding_txo: Option, + ) { + let Some(record) = self.record_for_channel(user_channel_id) else { + return; + }; + let id = record.id(); + let has_payment = self.has_payment(&record); + let Some(intent) = record.splice_intent() else { + return; + }; + // Only clear an intent that predates the locked funding. An intent whose pre-splice outpoint + // still matches the newly locked funding was created after this lock and is still pending. + let clear = match funding_txo { + Some(funding_txo) => { + intent.pre_splice_funding_txo.into_bitcoin_outpoint() != funding_txo + }, + None => false, + }; + if clear { + self.clear_intent(id, has_payment).await; + } + } + + /// Clears any splice intent for a closed channel, as there is nothing left to splice. + pub(crate) async fn on_channel_closed(&self, user_channel_id: UserChannelId) { + if let Some(record) = self.record_for_channel(user_channel_id) { + let has_payment = self.has_payment(&record); + self.clear_intent(record.id(), has_payment).await; + } + } + + /// Returns the pending record carrying a splice intent for the given channel, if any. + fn record_for_channel(&self, user_channel_id: UserChannelId) -> Option { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|i| i.user_channel_id == user_channel_id) + }) + .into_iter() + .next() + } + + /// Whether a classified funding payment exists for this record: its own details, or a + /// still-`Pending` record in the payment store — the latter without the former means a crash + /// landed between classification's two store writes, so the entry is still pre-broadcast + /// while the payment record already exists. + fn has_payment(&self, record: &PendingPaymentDetails) -> bool { + record.details().is_some() + || self + .payment_store + .get(&record.id()) + .is_some_and(|details| details.status == PaymentStatus::Pending) + } + + /// Builds a fresh contribution from the parameters of the originating API call, mirroring the + /// corresponding [`Node`] method. + /// + /// [`Node`]: crate::Node + async fn rebuild_contribution( + &self, channel_id: &ChannelId, counterparty_node_id: &PublicKey, kind: &SpliceKind, + ) -> Result { + let template = self + .channel_manager + .splice_channel(channel_id, counterparty_node_id) + .map_err(|_| Error::ChannelSplicingFailed)?; + + let est_feerate = self.fee_estimator.estimate_fee_rate(ConfirmationTarget::ChannelFunding); + let max_feerate = max_funding_feerate(est_feerate); + let feerate = match template.min_rbf_feerate() { + Some(min_rbf_feerate) if min_rbf_feerate <= max_feerate => { + est_feerate.max(min_rbf_feerate) + }, + _ => est_feerate, + }; + + match kind { + SpliceKind::In { amount_sats } => template + .splice_in( + Amount::from_sat(*amount_sats), + feerate, + max_feerate, + Arc::clone(&self.wallet), + ) + .await + .map_err(|_| Error::ChannelSplicingFailed), + SpliceKind::Out { outputs } => template + .splice_out(outputs.clone(), feerate, max_feerate) + .map_err(|_| Error::ChannelSplicingFailed), + SpliceKind::Rbf {} => template + .rbf_prior_contribution(None, max_feerate, Arc::clone(&self.wallet)) + .await + .map_err(|_| Error::ChannelSplicingFailed), + } + } +} + +/// The replacement for a pending record whose splice intent is being dropped. A tracked record +/// keeps its payment details with just the intent cleared. A pre-broadcast record whose payment +/// was classified but never mirrored into the pending store — a crash between classification's +/// two store writes — is promoted so the payment keeps graduating and its txids stay mapped; a +/// payment no longer `Pending` graduated already and must not be re-indexed, so its entry is +/// left alone for removal. +fn record_with_intent_cleared( + existing: Option<&PendingPaymentDetails>, recorded: Option, +) -> Option { + match existing { + Some(PendingPaymentDetails::PendingSplice { .. }) => recorded + .filter(|details| details.status == PaymentStatus::Pending) + .map(|details| PendingPaymentDetails::tracked(details, Vec::new(), Vec::new(), None)), + Some(tracked @ PendingPaymentDetails::Tracked { .. }) => { + let update = PendingPaymentDetailsUpdate { + id: tracked.id(), + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(None), + }; + let mut updated = tracked.clone(); + updated.update(update).then_some(updated) + }, + None => None, + } +} + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use bitcoin::hashes::Hash; + use bitcoin::Txid; + use lightning::chain::transaction::OutPoint; + + use super::*; + use crate::payment::pending_payment_store::{test_funding_contribution, PendingPaymentDetails}; + use crate::payment::store::{ConfirmationStatus, PaymentDetails, PaymentKind}; + use crate::payment::{PaymentDirection, PaymentStatus}; + use crate::types::UserChannelId; + + #[test] + fn decide_retry_matrix() { + use NegotiationFailureReason::*; + + // A non-retriable reason gives up regardless of attempts. + assert_eq!(decide_retry(&LocallyCanceled, 0), RetryDecision::Abandon); + // Retriable, but the resubmission budget is exhausted -> give up. + assert_eq!(decide_retry(&PeerDisconnected, MAX_SPLICE_ATTEMPTS), RetryDecision::Abandon); + // Transient failures resubmit the stored contribution. + assert_eq!(decide_retry(&PeerDisconnected, 0), RetryDecision::ResubmitStored); + assert_eq!(decide_retry(&Unknown, MAX_SPLICE_ATTEMPTS - 1), RetryDecision::ResubmitStored); + // A stale contribution is rebuilt from the original parameters. + assert_eq!(decide_retry(&FeeRateTooLow, 0), RetryDecision::Rebuild); + assert_eq!(decide_retry(&ContributionInvalid, 0), RetryDecision::Rebuild); + } + + fn test_intent() -> SpliceIntent { + SpliceIntent { + user_channel_id: UserChannelId(42), + counterparty_node_id: PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(), + channel_id: ChannelId([7u8; 32]), + pre_splice_funding_txo: OutPoint { txid: Txid::from_byte_array([3u8; 32]), index: 0 }, + contribution: test_funding_contribution(), + kind: SpliceKind::Rbf {}, + attempts: 0, + } + } + + fn payment_details(id: PaymentId, status: PaymentStatus) -> PaymentDetails { + PaymentDetails::new( + id, + PaymentKind::Onchain { + txid: Txid::from_byte_array([1u8; 32]), + status: ConfirmationStatus::Unconfirmed, + tx_type: None, + }, + Some(1_000_000), + Some(500), + PaymentDirection::Outbound, + status, + ) + } + + /// A crash between classification's two store writes leaves the pending entry pre-broadcast + /// while the classified payment record exists. Dropping the intent must promote the entry + /// rather than remove it, so the payment keeps graduating and its txids stay mapped. + #[test] + fn intent_clearing_promotes_a_pre_broadcast_record_over_a_classified_payment() { + let id = PaymentId([9u8; 32]); + let existing = PendingPaymentDetails::pending_splice(id, test_intent()); + let recorded = payment_details(id, PaymentStatus::Pending); + + let replacement = record_with_intent_cleared(Some(&existing), Some(recorded.clone())); + let replacement = replacement.expect("the entry must be promoted, not removed"); + assert_eq!(replacement.details(), Some(&recorded)); + assert!(replacement.splice_intent().is_none()); + } + + /// A payment that already advanced beyond `Pending` graduated and lost its pending entry; + /// promotion must not re-index it. + #[test] + fn intent_clearing_does_not_reindex_an_advanced_payment() { + let id = PaymentId([9u8; 32]); + let existing = PendingPaymentDetails::pending_splice(id, test_intent()); + let recorded = payment_details(id, PaymentStatus::Succeeded); + assert!(record_with_intent_cleared(Some(&existing), Some(recorded)).is_none()); + } + + /// A tracked record keeps its payment details; only the intent is cleared. + #[test] + fn intent_clearing_keeps_a_tracked_record() { + let id = PaymentId([9u8; 32]); + let details = payment_details(id, PaymentStatus::Pending); + let existing = PendingPaymentDetails::tracked( + details.clone(), + Vec::new(), + Vec::new(), + Some(test_intent()), + ); + + let replacement = record_with_intent_cleared(Some(&existing), Some(details.clone())); + let replacement = replacement.expect("the entry must survive with its intent cleared"); + assert_eq!(replacement.details(), Some(&details)); + assert!(replacement.splice_intent().is_none()); + } +} diff --git a/src/event.rs b/src/event.rs index be54969c7..8149d99f9 100644 --- a/src/event.rs +++ b/src/event.rs @@ -34,6 +34,7 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_liquidity::lsps2::utils::compute_opening_fee; use lightning_types::payment::{PaymentHash, PaymentPreimage}; +use crate::channel::SpliceRetrier; use crate::config::{may_announce_channel, Config, PEER_RECONNECTION_INTERVAL}; use crate::connection::ConnectionManager; use crate::data_store::DataStoreUpdateResult; @@ -295,9 +296,13 @@ pub enum Event { /// The outpoint of the channel's splice funding transaction. new_funding_txo: OutPoint, }, - /// A channel splice negotiation round with local inputs or outputs has failed. + /// A channel splice has failed and is no longer being pursued. /// /// This event is not emitted when only the counterparty contributes to a splice. + /// + /// A recoverable failure of a user-initiated splice (e.g. the peer disconnecting + /// mid-negotiation) is retried automatically, including across restarts; this event is emitted + /// only once the splice is given up on. SpliceNegotiationFailed { /// The `channel_id` of the channel. channel_id: ChannelId, @@ -557,6 +562,7 @@ where onion_messenger: Arc, om_mailbox: Option>, prober: Option>, + splice_retrier: Arc>, runtime: Arc, logger: L, config: Arc, @@ -575,7 +581,8 @@ where peer_store: Arc>, keys_manager: Arc, static_invoice_store: Option, onion_messenger: Arc, om_mailbox: Option>, prober: Option>, - runtime: Arc, logger: L, config: Arc, + splice_retrier: Arc>, runtime: Arc, logger: L, + config: Arc, ) -> Self { Self { event_queue, @@ -593,6 +600,7 @@ where onion_messenger, om_mailbox, prober, + splice_retrier, runtime, logger, config, @@ -1896,6 +1904,10 @@ where .handle_channel_ready(user_channel_id, &channel_id, &counterparty_node_id) .await; + self.splice_retrier + .on_channel_ready(UserChannelId(user_channel_id), funding_txo) + .await; + let event = Event::ChannelReady { channel_id, user_channel_id: UserChannelId(user_channel_id), @@ -1919,6 +1931,8 @@ where } => { log_info!(self.logger, "Channel {} closed due to: {}", channel_id, reason); + self.splice_retrier.on_channel_closed(UserChannelId(user_channel_id)).await; + // `counterparty_node_id` has been set on every `ChannelClosed` since LDK 0.0.117. let counterparty_node_id = counterparty_node_id .expect("counterparty_node_id is always set since LDK 0.0.117"); @@ -2230,15 +2244,27 @@ where channel_id, user_channel_id, counterparty_node_id, - .. + reason, + contribution, } => { log_info!( self.logger, - "Channel {} with counterparty {} splice negotiation failed", + "Channel {} with counterparty {} splice negotiation failed: {}", channel_id, counterparty_node_id, + reason, ); + // A user-initiated splice is retried automatically, including across restarts; + // surface the failure only once it is given up on. + let surface = self + .splice_retrier + .on_negotiation_failed(UserChannelId(user_channel_id), reason, contribution) + .await; + if !surface { + return Ok(()); + } + let event = Event::SpliceNegotiationFailed { channel_id, user_channel_id: UserChannelId(user_channel_id), diff --git a/src/lib.rs b/src/lib.rs index 14ee734a3..52bef135c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -83,6 +83,7 @@ mod balance; mod builder; mod chain; +mod channel; pub mod config; mod connection; mod data_store; @@ -128,12 +129,14 @@ pub use builder::BuildError; #[cfg(not(feature = "uniffi"))] pub use builder::NodeBuilder as Builder; use chain::ChainSource; +use channel::SpliceRetrier; use config::{ default_user_config, may_announce_channel, AsyncPaymentsRole, ChannelConfig, Config, LNURL_AUTH_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL, PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL, }; use connection::ConnectionManager; +use data_store::StorableObject; pub use error::Error as NodeError; use error::Error; pub use event::Event; @@ -153,6 +156,7 @@ use lightning::ln::chan_utils::FUNDING_TRANSACTION_WITNESS_WEIGHT; use lightning::ln::channel_state::ChannelDetails as LdkChannelDetails; pub use lightning::ln::channel_state::ChannelShutdownState; use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; use lightning::ln::msgs::{BaseMessageHandler, SocketAddress}; use lightning::ln::peer_handler::CustomMessageHandler; use lightning::routing::gossip::NodeAlias; @@ -171,6 +175,9 @@ use lnurl_auth::LnurlAuth; use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use payment::asynchronous::om_mailbox::OnionMessageMailbox; use payment::asynchronous::static_invoice_store::StaticInvoiceStore; +use payment::pending_payment_store::{ + PendingPaymentDetails, PendingPaymentDetailsUpdate, SpliceIntent, SpliceKind, +}; use payment::{ Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment, UnifiedPayment, @@ -183,8 +190,8 @@ use runtime::Runtime; pub use tokio; use types::{ Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph, - HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper, - Wallet, + HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore, + Router, Scorer, Sweeper, Wallet, }; pub use types::{ ChannelCounterparty, ChannelDetails, CustomTlvRecord, PeerDetails, ReserveType, UserChannelId, @@ -265,6 +272,7 @@ pub struct Node { scorer: Arc>, peer_store: Arc>>, payment_store: Arc, + pending_payment_store: Arc, lnurl_auth: Arc, is_running: Arc>, node_metrics: Arc, @@ -661,6 +669,16 @@ impl Node { None }; + let splice_retrier = Arc::new(SpliceRetrier::new( + Arc::clone(&self.channel_manager), + Arc::clone(&self.wallet), + Arc::clone(&self.fee_estimator), + Arc::clone(&self.pending_payment_store), + Arc::clone(&self.payment_store), + Arc::clone(&self.event_queue), + Arc::clone(&self.logger), + )); + let event_handler = Arc::new(EventHandler::new( Arc::clone(&self.event_queue), Arc::clone(&self.wallet), @@ -677,6 +695,7 @@ impl Node { Arc::clone(&self.onion_messenger), self.om_mailbox.clone(), self.prober.clone(), + Arc::clone(&splice_retrier), Arc::clone(&self.runtime), Arc::clone(&self.logger), Arc::clone(&self.config), @@ -689,6 +708,11 @@ impl Node { }); } + // Resubmit any persisted splice intents that LDK dropped before durably recording them. + self.runtime.spawn_background_task(async move { + splice_retrier.reconcile().await; + }); + // Setup background processing let background_persister = Arc::clone(&self.kv_store); let background_event_handler = Arc::clone(&event_handler); @@ -1665,6 +1689,87 @@ impl Node { ) } + /// Persists a splice intent before its contribution is handed to LDK, so the splice can be + /// resubmitted if LDK drops it before durably recording it (a restart, or a disconnect + /// mid-negotiation). Must be called before `funding_contributed` so a crash in between is also + /// covered. + /// + /// Reuses the channel's existing splice intent record when one is present -- so a splice and its + /// later fee bumps share one [`PaymentId`] and at most one intent ever exists per channel, which + /// [`Wallet::find_splice_payment_id`] and the retrier rely on -- otherwise generates a fresh id. + /// Returns the id and, for restoring on failure, `None` when a fresh record was created or + /// `Some(prior)` when an existing record's intent was replaced. + fn persist_splice_intent( + &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, + channel_details: &LdkChannelDetails, contribution: FundingContribution, kind: SpliceKind, + ) -> Result<(PaymentId, Option>), Error> { + let pre_splice_funding_txo = channel_details.funding_txo.ok_or_else(|| { + log_error!(self.logger, "Failed to splice channel: channel not yet ready"); + Error::ChannelSplicingFailed + })?; + let intent = SpliceIntent { + user_channel_id: *user_channel_id, + counterparty_node_id, + channel_id: channel_details.channel_id, + pre_splice_funding_txo, + contribution, + kind, + attempts: 0, + }; + let existing = self + .pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|i| i.user_channel_id == *user_channel_id) + }) + .into_iter() + .next(); + match existing { + Some(record) => { + let payment_id = record.id(); + let prior = record.splice_intent().cloned(); + self.runtime.block_on(self.pending_payment_store.update( + PendingPaymentDetailsUpdate { + id: payment_id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent)), + }, + ))?; + Ok((payment_id, Some(prior))) + }, + None => { + let payment_id = PaymentId(self.keys_manager.get_secure_random_bytes()); + self.runtime.block_on( + self.pending_payment_store + .insert(PendingPaymentDetails::pending_splice(payment_id, intent)), + )?; + Ok((payment_id, None)) + }, + } + } + + /// Undoes a splice intent persisted for an originating call whose `funding_contributed` then + /// failed: restores an existing record's prior intent, or removes a freshly created record. + fn discard_splice_intent(&self, payment_id: &PaymentId, restore: Option>) { + match restore { + Some(prior) => { + let _ = self.runtime.block_on(self.pending_payment_store.update( + PendingPaymentDetailsUpdate { + id: *payment_id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(prior), + }, + )); + }, + None => { + let _ = self.runtime.block_on(self.pending_payment_store.remove(payment_id)); + }, + } + } + fn splice_in_inner( &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, splice_amount_sats: FundingAmount, @@ -1773,6 +1878,14 @@ impl Node { Error::ChannelSplicingFailed })?; + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::In { amount_sats: splice_amount_sats }, + )?; + self.channel_manager .funding_contributed( &channel_details.channel_id, @@ -1782,6 +1895,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to splice channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { @@ -1801,6 +1915,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-in will be marked as an outbound payment, but @@ -1825,6 +1943,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-in will be marked as an outbound payment, but @@ -1841,6 +1963,10 @@ impl Node { /// it. Once negotiation with the counterparty is complete, the channel remains operational /// while waiting for a new funding transaction to confirm. /// + /// The splice is retried automatically, including across restarts, until it either completes or + /// fails for a reason retrying cannot address, at which point [`Event::SpliceNegotiationFailed`] + /// is emitted. + /// /// # Experimental API /// /// This API is experimental. Currently, a splice-out will be marked as an inbound payment if @@ -1897,11 +2023,20 @@ impl Node { value: Amount::from_sat(splice_amount_sats), script_pubkey: address.script_pubkey(), }]; - let contribution = - funding_template.splice_out(outputs, feerate, max_feerate).map_err(|e| { - log_error!(self.logger, "Failed to splice channel: {}", e); - Error::ChannelSplicingFailed - })?; + let contribution = funding_template + .splice_out(outputs.clone(), feerate, max_feerate) + .map_err(|e| { + log_error!(self.logger, "Failed to splice channel: {}", e); + Error::ChannelSplicingFailed + })?; + + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::Out { outputs }, + )?; self.channel_manager .funding_contributed( @@ -1912,6 +2047,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to splice channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { @@ -1928,6 +2064,10 @@ impl Node { /// Fee-bumps the pending splice on a channel by replacing its in-flight funding transaction /// (RBF). The splice's amount and destination are preserved; only the fee rate is raised. /// Errors if the channel has no pending splice to bump. + /// + /// The fee bump is retried automatically, including across restarts, until it either completes + /// or fails for a reason retrying cannot address, at which point + /// [`Event::SpliceNegotiationFailed`] is emitted. pub fn bump_channel_funding_fee( &self, user_channel_id: &UserChannelId, counterparty_node_id: PublicKey, ) -> Result<(), Error> { @@ -1974,6 +2114,14 @@ impl Node { Error::ChannelSplicingFailed })?; + let (payment_id, restore) = self.persist_splice_intent( + user_channel_id, + counterparty_node_id, + channel_details, + contribution.clone(), + SpliceKind::Rbf {}, + )?; + self.channel_manager .funding_contributed( &channel_details.channel_id, @@ -1983,6 +2131,7 @@ impl Node { ) .map_err(|e| { log_error!(self.logger, "Failed to RBF channel: {:?}", e); + self.discard_splice_intent(&payment_id, restore); Error::ChannelSplicingFailed }) } else { diff --git a/src/payment/pending_payment_store.rs b/src/payment/pending_payment_store.rs index 30a113537..4994a1f11 100644 --- a/src/payment/pending_payment_store.rs +++ b/src/payment/pending_payment_store.rs @@ -5,13 +5,22 @@ // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in // accordance with one or both of these licenses. -use bitcoin::Txid; -use lightning::impl_writeable_tlv_based; +use bitcoin::secp256k1::PublicKey; +use bitcoin::{TxOut, Txid}; +use lightning::chain::transaction::OutPoint; use lightning::ln::channelmanager::PaymentId; +use lightning::ln::funding::FundingContribution; +use lightning::ln::types::ChannelId; +use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use crate::data_store::{StorableObject, StorableObjectUpdate}; use crate::payment::store::PaymentDetailsUpdate; use crate::payment::{PaymentDetails, PaymentKind}; +use crate::types::UserChannelId; + +/// The number of times a splice intent is resubmitted to LDK before it is abandoned and the +/// failure is surfaced to the user. +pub(crate) const MAX_SPLICE_ATTEMPTS: u8 = 3; /// One candidate transaction in an interactive-funding (splice) RBF history, holding this node's /// share of the funding amount and fee for that candidate. Both are `None` for a candidate this @@ -36,37 +45,169 @@ impl_writeable_tlv_based!(FundingTxCandidate, { (4, fee_paid_msat, option), }); -/// Represents a pending payment +/// The parameters of the API call that initiated a splice, used to rebuild a fresh contribution +/// when the stored one has become stale (e.g. its feerate is no longer sufficient). #[derive(Clone, Debug, PartialEq, Eq)] -pub struct PendingPaymentDetails { - /// The full payment details - pub details: PaymentDetails, - /// Transaction IDs that have replaced or conflict with this payment. - pub conflicting_txids: Vec, - /// For interactive funding (splices), this node's per-candidate funding figures across the - /// RBF history, keyed by each candidate's txid. Empty for non-funding payments and for - /// records written before per-candidate tracking existed. - pub(crate) candidates: Vec, +pub(crate) enum SpliceKind { + /// [`Node::splice_in`] with a resolved amount. + /// + /// [`Node::splice_in`]: crate::Node::splice_in + In { amount_sats: u64 }, + /// [`Node::splice_out`] to the given outputs. + /// + /// [`Node::splice_out`]: crate::Node::splice_out + Out { outputs: Vec }, + /// [`Node::bump_channel_funding_fee`] of a pending splice. + /// + /// [`Node::bump_channel_funding_fee`]: crate::Node::bump_channel_funding_fee + Rbf {}, +} + +impl_writeable_tlv_based_enum!(SpliceKind, + (0, In) => { + (0, amount_sats, required), + }, + (2, Out) => { + (0, outputs, required_vec), + }, + (4, Rbf) => {}, +); + +/// A user-initiated splice that has been handed to LDK but is not yet guaranteed to survive a +/// restart. LDK only persists a splice once its negotiation reaches `AwaitingSignatures`, and it +/// abandons an in-progress negotiation whenever the peer disconnects (which includes stopping the +/// node). Until the new funding transaction locks we keep enough state to resubmit the splice +/// ourselves. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct SpliceIntent { + /// The channel's local identifier, carried so the retrier can address the splice without a + /// separate store keyed by it. + pub user_channel_id: UserChannelId, + /// The channel counterparty. + pub counterparty_node_id: PublicKey, + /// The channel being spliced. + pub channel_id: ChannelId, + /// The channel's funding outpoint when the splice was initiated. It only changes once a splice + /// locks, so a mismatch with the channel's current funding outpoint means the splice (or a + /// replacement) completed and there is nothing left to resubmit. + pub pre_splice_funding_txo: OutPoint, + /// The contribution handed to [`ChannelManager::funding_contributed`], resubmitted verbatim. + /// + /// [`ChannelManager::funding_contributed`]: lightning::ln::channelmanager::ChannelManager::funding_contributed + pub contribution: FundingContribution, + /// The parameters of the originating API call, used to rebuild a fresh contribution when the + /// stored one has become stale. + pub kind: SpliceKind, + /// The number of times the contribution has been resubmitted to LDK after the originating API + /// call handed it off. + pub attempts: u8, +} + +impl_writeable_tlv_based!(SpliceIntent, { + (0, user_channel_id, required), + (2, counterparty_node_id, required), + (4, channel_id, required), + (6, pre_splice_funding_txo, required), + (8, contribution, required), + (10, kind, required), + (12, attempts, required), +}); + +/// A pending payment tracked by LDK Node, keyed by [`PaymentId`]. +/// +/// A user-initiated splice is persisted as a [`PendingSplice`] before its contribution is handed +/// to LDK — at which point no funding transaction, and therefore no [`PaymentDetails`], exists yet. +/// Once the splice is broadcast and classified it becomes a [`Tracked`] payment carrying the real +/// [`PaymentDetails`], while retaining its [`SpliceIntent`] until the splice locks. +/// +/// [`PendingSplice`]: Self::PendingSplice +/// [`Tracked`]: Self::Tracked +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum PendingPaymentDetails { + /// A user-initiated splice persisted before hand-off to LDK; no funding transaction exists yet. + /// Keyed by the generated [`PaymentId`]; never mirrored into the payment store. + PendingSplice { id: PaymentId, intent: SpliceIntent }, + /// A pending payment tracked toward confirmation, optionally still carrying a live splice + /// intent to resubmit until the splice locks. + Tracked { + /// The full payment details. + details: PaymentDetails, + /// Transaction IDs that have replaced or conflict with this payment. + conflicting_txids: Vec, + /// For interactive funding (splices), this node's per-candidate funding figures across the + /// RBF history, keyed by each candidate's txid. Empty for non-funding payments. + candidates: Vec, + /// The splice intent to resubmit if LDK drops the splice before it locks, or `None` for a + /// non-splice payment or a splice that has locked. + splice_intent: Option, + }, } impl PendingPaymentDetails { pub(crate) fn new( details: PaymentDetails, conflicting_txids: Vec, candidates: Vec, ) -> Self { - Self { details, conflicting_txids, candidates } + Self::tracked(details, conflicting_txids, candidates, None) + } + + pub(crate) fn tracked( + details: PaymentDetails, conflicting_txids: Vec, candidates: Vec, + splice_intent: Option, + ) -> Self { + Self::Tracked { details, conflicting_txids, candidates, splice_intent } + } + + pub(crate) fn pending_splice(id: PaymentId, intent: SpliceIntent) -> Self { + Self::PendingSplice { id, intent } + } + + /// The full payment details, or `None` for a splice not yet broadcast. + pub(crate) fn details(&self) -> Option<&PaymentDetails> { + match self { + Self::PendingSplice { .. } => None, + Self::Tracked { details, .. } => Some(details), + } + } + + /// Transaction IDs that have replaced or conflict with this payment. + pub(crate) fn conflicting_txids(&self) -> &[Txid] { + match self { + Self::PendingSplice { .. } => &[], + Self::Tracked { conflicting_txids, .. } => conflicting_txids, + } + } + + /// The splice intent this record carries, if it is a splice that has not yet locked. + pub(crate) fn splice_intent(&self) -> Option<&SpliceIntent> { + match self { + Self::PendingSplice { intent, .. } => Some(intent), + Self::Tracked { splice_intent, .. } => splice_intent.as_ref(), + } } /// Returns this node's recorded funding figures for the candidate with the given txid, if any. pub(crate) fn candidate(&self, txid: Txid) -> Option<&FundingTxCandidate> { - self.candidates.iter().find(|candidate| candidate.txid == txid) + match self { + Self::PendingSplice { .. } => None, + Self::Tracked { candidates, .. } => { + candidates.iter().find(|candidate| candidate.txid == txid) + }, + } } } -impl_writeable_tlv_based!(PendingPaymentDetails, { - (0, details, required), - (2, conflicting_txids, optional_vec), - (4, candidates, optional_vec), -}); +impl_writeable_tlv_based_enum!(PendingPaymentDetails, + (0, PendingSplice) => { + (0, id, required), + (2, intent, required), + }, + (2, Tracked) => { + (0, details, required), + (2, conflicting_txids, optional_vec), + (4, candidates, optional_vec), + (6, splice_intent, option), + }, +); #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct PendingPaymentDetailsUpdate { @@ -74,6 +215,11 @@ pub(crate) struct PendingPaymentDetailsUpdate { pub payment_update: Option, pub conflicting_txids: Option>, pub candidates: Vec, + /// The splice intent to set (`Some(Some(..))`) or clear (`Some(None)`), or `None` to leave it + /// unchanged. Setting it on a [`PendingPaymentDetails::PendingSplice`] replaces the intent (e.g. + /// to bump the retry attempt count); clearing a pre-broadcast splice is done by removing the + /// record, not through this field. + pub splice_intent: Option>, } impl StorableObject for PendingPaymentDetails { @@ -81,38 +227,64 @@ impl StorableObject for PendingPaymentDetails { type Update = PendingPaymentDetailsUpdate; fn id(&self) -> Self::Id { - self.details.id + match self { + Self::PendingSplice { id, .. } => *id, + Self::Tracked { details, .. } => details.id, + } } fn update(&mut self, update: Self::Update) -> bool { - let mut updated = false; - - // Update the underlying payment details if present - if let Some(payment_update) = update.payment_update { - updated |= self.details.update(payment_update); - } - - if let Some(new_conflicting_txids) = update.conflicting_txids { - if self.conflicting_txids != new_conflicting_txids { - self.conflicting_txids = new_conflicting_txids; - updated = true; - } - } - - if let PaymentKind::Onchain { txid, .. } = &self.details.kind { - let conflicts_len = self.conflicting_txids.len(); - self.conflicting_txids.retain(|conflicting_txid| conflicting_txid != txid); - updated |= self.conflicting_txids.len() != conflicts_len; - } - - // Each classify passes the complete candidate history, so a non-empty update replaces the - // stored list. An empty update (e.g. a non-funding payment) leaves it untouched. - if !update.candidates.is_empty() && self.candidates != update.candidates { - self.candidates = update.candidates; - updated = true; + match self { + Self::PendingSplice { intent, .. } => { + // A pre-broadcast record only carries a splice intent; the only meaningful update is + // replacing that intent. Clearing it is done by removing the record. + if let Some(Some(new_intent)) = update.splice_intent { + if *intent != new_intent { + *intent = new_intent; + return true; + } + } + false + }, + Self::Tracked { details, conflicting_txids, candidates, splice_intent } => { + let mut updated = false; + + // Update the underlying payment details if present + if let Some(payment_update) = update.payment_update { + updated |= details.update(payment_update); + } + + if let Some(new_conflicting_txids) = update.conflicting_txids { + if *conflicting_txids != new_conflicting_txids { + *conflicting_txids = new_conflicting_txids; + updated = true; + } + } + + if let PaymentKind::Onchain { txid, .. } = &details.kind { + let conflicts_len = conflicting_txids.len(); + conflicting_txids.retain(|conflicting_txid| conflicting_txid != txid); + updated |= conflicting_txids.len() != conflicts_len; + } + + // Each classify passes the complete candidate history, so a non-empty update + // replaces the stored list. An empty update (e.g. a non-funding payment) leaves it + // untouched. + if !update.candidates.is_empty() && *candidates != update.candidates { + *candidates = update.candidates; + updated = true; + } + + if let Some(new_splice_intent) = update.splice_intent { + if *splice_intent != new_splice_intent { + *splice_intent = new_splice_intent; + updated = true; + } + } + + updated + }, } - - updated } fn to_update(&self) -> Self::Update { @@ -128,23 +300,62 @@ impl StorableObjectUpdate for PendingPaymentDetailsUpdate impl From<&PendingPaymentDetails> for PendingPaymentDetailsUpdate { fn from(value: &PendingPaymentDetails) -> Self { - let conflicting_txids = if value.conflicting_txids.is_empty() { - None - } else { - Some(value.conflicting_txids.clone()) - }; - Self { - id: value.id(), - payment_update: Some(value.details.to_update()), - conflicting_txids, - candidates: value.candidates.clone(), + match value { + PendingPaymentDetails::PendingSplice { id, intent } => Self { + id: *id, + payment_update: None, + conflicting_txids: None, + candidates: Vec::new(), + splice_intent: Some(Some(intent.clone())), + }, + PendingPaymentDetails::Tracked { + details, + conflicting_txids, + candidates, + splice_intent, + } => { + let conflicting_txids = if conflicting_txids.is_empty() { + None + } else { + Some(conflicting_txids.clone()) + }; + // Leave the splice intent unchanged: it is owned by the splice entry points and the + // retrier, never by a payment-tracking merge. Emitting the current value here would + // let an `insert_or_update` of a payment record (e.g. from wallet sync, built without + // an intent) clobber a live intent to `None`. + let _ = splice_intent; + Self { + id: details.id, + payment_update: Some(details.to_update()), + conflicting_txids, + candidates: candidates.clone(), + splice_intent: None, + } + }, } } } +/// Builds a [`FundingContribution`] for tests through its `Readable` impl — the only path open +/// outside `rust-lightning`, which keeps its builder private. The length-prefixed stream holds +/// just the required TLV records: estimated fee, feerate, max feerate, and the is-splice flag. +#[cfg(test)] +pub(crate) fn test_funding_contribution() -> FundingContribution { + let tlv_bytes = [ + 33u8, // BigSize length prefix over the TLV records below + 1, 8, 0, 0, 0, 0, 0, 0, 0, 0, // (1, estimated_fee: 0 sat) + 9, 8, 0, 0, 0, 0, 0, 0, 0, 253, // (9, feerate: 253 sat/kwu) + 11, 8, 0, 0, 0, 0, 0, 0, 0, 253, // (11, max_feerate: 253 sat/kwu) + 13, 1, 1, // (13, is_splice: true) + ]; + lightning::util::ser::Readable::read(&mut &tlv_bytes[..]) + .expect("hand-built TLV stream must decode") +} + #[cfg(test)] mod tests { use bitcoin::hashes::Hash; + use lightning::util::ser::{Readable, Writeable}; use super::*; use crate::payment::store::ConfirmationStatus; @@ -237,7 +448,7 @@ mod tests { assert!(pending_payment.update(update)); assert_eq!( - pending_payment.conflicting_txids, + pending_payment.conflicting_txids(), Vec::::new(), "current txid must not remain in its own conflict list" ); @@ -289,7 +500,7 @@ mod tests { assert!(downgraded.update(full_update)); assert!( matches!( - downgraded.details.kind, + downgraded.details().expect("tracked").kind, PaymentKind::Onchain { status: ConfirmationStatus::Unconfirmed, .. } ), "a full merge of a fresh classification downgrades a mirrored confirmation", @@ -304,17 +515,56 @@ mod tests { payment_update: Some(PaymentDetailsUpdate::funding_reclassification(fresh)), conflicting_txids: None, candidates: candidates.clone(), + splice_intent: None, }; assert!(merged.update(narrow_update)); + let merged_details = merged.details().expect("tracked"); assert!( matches!( - merged.details.kind, + merged_details.kind, PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { .. }, .. } ), "a narrow classification update must not downgrade a mirrored confirmation", ); - assert_eq!(merged.candidates, candidates); - assert_eq!(merged.details.amount_msat, Some(1_000)); - assert_eq!(merged.details.fee_paid_msat, Some(100)); + assert_eq!(merged.candidate(txid), Some(&candidates[0])); + assert_eq!(merged_details.amount_msat, Some(1_000)); + assert_eq!(merged_details.fee_paid_msat, Some(100)); + } + + #[test] + fn splice_kind_round_trips() { + for kind in [ + SpliceKind::In { amount_sats: 500_000 }, + SpliceKind::Out { + outputs: vec![TxOut { + value: bitcoin::Amount::from_sat(400_000), + script_pubkey: bitcoin::ScriptBuf::new(), + }], + }, + SpliceKind::Rbf {}, + ] { + let encoded = kind.encode(); + let decoded = SpliceKind::read(&mut &encoded[..]).unwrap(); + assert_eq!(kind, decoded); + } + } + + #[test] + fn tracked_payment_round_trips() { + // A `PendingSplice` record round-trips through the restart integration tests, which persist a + // real `FundingContribution`; here we cover the `Tracked` variant and its enum discriminant. + let payment_id = PaymentId([7u8; 32]); + let txid = Txid::from_byte_array([8u8; 32]); + let record = PendingPaymentDetails::new( + pending_onchain_payment(payment_id, txid), + vec![Txid::from_byte_array([9u8; 32])], + vec![FundingTxCandidate { txid, amount_msat: Some(1_000), fee_paid_msat: Some(100) }], + ); + + let encoded = record.encode(); + let decoded = PendingPaymentDetails::read(&mut &encoded[..]).unwrap(); + assert_eq!(record, decoded); + assert_eq!(decoded.id(), payment_id); + assert!(decoded.details().is_some()); } } diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index df95c11ec..2f81c900e 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -57,7 +57,7 @@ use crate::config::{Config, ADDRESS_POOL_SIZE}; use crate::data_store::StorableObject; use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; -use crate::payment::pending_payment_store::PendingPaymentDetailsUpdate; +use crate::payment::pending_payment_store::{PendingPaymentDetailsUpdate, SpliceKind}; use crate::payment::store::{ConfirmationStatus, PaymentDetailsUpdate}; use crate::payment::{ FundingTxCandidate, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, @@ -342,7 +342,7 @@ impl Wallet { let guard = self.funding_payment_update_lock.lock().await; let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -372,33 +372,38 @@ impl Wallet { self.payment_store.insert_or_update(payment.clone()).await?; if payment_status == PaymentStatus::Pending { - let pending_payment = - self.create_pending_payment_from_tx(payment, Vec::new()); - - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; } }, WalletEvent::ChainTipChanged { new_tip, .. } => { let pending_payments: Vec = - self.pending_payment_store.list_filter(|p| { - debug_assert!( - p.details.status == PaymentStatus::Pending, - "Non-pending payment {:?} found in pending store", - p.details.id, - ); - p.details.status == PaymentStatus::Pending - && matches!(p.details.kind, PaymentKind::Onchain { .. }) + self.pending_payment_store.list_filter(|p| match p.details() { + // A pre-broadcast splice intent carries no payment yet and cannot graduate. + None => false, + Some(details) => { + debug_assert!( + details.status == PaymentStatus::Pending, + "Non-pending payment {:?} found in pending store", + details.id, + ); + details.status == PaymentStatus::Pending + && matches!(details.kind, PaymentKind::Onchain { .. }) + }, }); let mut unconfirmed_outbound_txids: Vec = Vec::new(); for payment in pending_payments { - match payment.details.kind { + // The filter admits only Tracked funding payments. + let PendingPaymentDetails::Tracked { details, .. } = payment else { + continue; + }; + match details.kind { PaymentKind::Onchain { status: ConfirmationStatus::Confirmed { height, .. }, .. } => { - let payment_id = payment.details.id; + let payment_id = details.id; if new_tip.height >= height + ANTI_REORG_DELAY - 1 { // Graduate from the live record, not the snapshot listed // above: a classification landing since then must not have @@ -441,7 +446,7 @@ impl Wallet { txid, status: ConfirmationStatus::Unconfirmed, .. - } if payment.details.direction == PaymentDirection::Outbound => { + } if details.direction == PaymentDirection::Outbound => { unconfirmed_outbound_txids.push(txid); }, _ => {}, @@ -480,7 +485,7 @@ impl Wallet { let guard = self.funding_payment_update_lock.lock().await; let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -506,10 +511,8 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ) }; - let pending_payment = - self.create_pending_payment_from_tx(payment.clone(), Vec::new()); - self.payment_store.insert_or_update(payment).await?; - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.payment_store.insert_or_update(payment.clone()).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; }, WalletEvent::TxReplaced { txid, conflicts, .. } => { // See `TxConfirmed`: id resolution and the writes below must not interleave @@ -544,10 +547,7 @@ impl Wallet { ); let payment = self.payment_store.get(&payment_id).ok_or(Error::InvalidPaymentId)?; - let pending_payment_details = - self.create_pending_payment_from_tx(payment, conflict_txids.clone()); - - self.pending_payment_store.insert_or_update(pending_payment_details).await?; + self.upsert_pending_payment(payment, conflict_txids).await?; }, WalletEvent::TxDropped { txid, tx } => { // See `TxConfirmed`: id resolution and the writes below must not interleave @@ -555,7 +555,7 @@ impl Wallet { let guard = self.funding_payment_update_lock.lock().await; let payment_id = self - .find_payment_by_txid(txid) + .find_payment_for_tx(&tx, txid) .unwrap_or_else(|| PaymentId(txid.to_byte_array())); if self @@ -581,10 +581,8 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ) }; - let pending_payment = - self.create_pending_payment_from_tx(payment.clone(), Vec::new()); - self.payment_store.insert_or_update(payment).await?; - self.pending_payment_store.insert_or_update(pending_payment).await?; + self.payment_store.insert_or_update(payment.clone()).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; }, _ => { continue; @@ -1576,7 +1574,14 @@ impl Wallet { return Ok(()); } - let payment_id = PaymentId(txid.to_byte_array()); + // A user-initiated splice's record is keyed by the PaymentId chosen at splice time, not + // by its funding txid — e.g. LDK re-broadcasts a promoted-but-unconfirmed 0conf splice + // through this generic funding path. Resolve to the existing record — or, when a crash + // left only the pre-broadcast intent behind, to the intent whose funding outpoint this + // transaction spends — so the broadcast merges into the splice's record rather than + // minting a duplicate under the txid-derived id. + let payment_id = + self.find_payment_for_tx(tx, txid).unwrap_or(PaymentId(txid.to_byte_array())); // A promoted-but-unconfirmed 0conf splice comes back through this generic path re-typed // and carrying wallet-view figures; `funding_reclassification_update` declines the @@ -1598,7 +1603,6 @@ impl Wallet { ); } } - let details = PaymentDetails::new( payment_id, PaymentKind::Onchain { @@ -1621,6 +1625,24 @@ impl Wallet { Ok(()) } + /// Returns the `PaymentId` of a user-initiated splice intent for one of the channels in + /// `candidate`, if any, so a classified splice adopts the id chosen at splice time rather than + /// deriving one from the first candidate's txid. A fee bump reuses the channel's existing intent, + /// so at most one in-flight intent matches and the first is unambiguous. + fn find_splice_payment_id(&self, candidate: &FundingCandidate) -> Option { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|intent| { + candidate.channels.iter().any(|channel| { + channel.channel_id == intent.channel_id + && channel.counterparty_node_id == intent.counterparty_node_id + }) + }) + }) + .first() + .map(|p| p.id()) + } + /// Records an interactive-funding broadcast (splice, or a V2 dual-funded open) as a pending /// on-chain payment, tagged with its transaction type. Amount and fee are this node's share, /// derived from the active candidate's contributions; broadcasts we didn't contribute to, or @@ -1671,9 +1693,18 @@ impl Wallet { return Ok(()); } - // Anchor the `PaymentId` to the first negotiated candidate so the record stays stable - // across RBF replacements. - let payment_id = PaymentId(first.txid.to_byte_array()); + // Adopt the `PaymentId` generated when the splice was initiated so its retry intent, funding + // payment, and candidate history share one record. If the intent is already gone (e.g. the + // splice locked before this classification ran), adopt the id of a record wallet sync + // created for any candidate rather than minting a divergent one. Fall back to the first + // negotiated candidate's txid for splices we did not originate (counterparty-initiated or + // V2 opens), which keeps that id stable across RBF replacements. + let payment_id = self + .find_splice_payment_id(active) + .or_else(|| { + candidates.iter().find_map(|candidate| self.find_payment_by_txid(candidate.txid)) + }) + .unwrap_or_else(|| PaymentId(first.txid.to_byte_array())); // Record every candidate's figures (`None` for any round we didn't contribute to, e.g. a // counterparty-initiated splice our `splice_in` later joined via RBF) so the confirmed @@ -1800,30 +1831,52 @@ impl Wallet { self.pending_payment_store .mutate(&id, |existing| { // The record was written above and payment records are never removed, so absence - // means the write failed out; fall back to the fresh details. + // means the write failed out; fall back to the fresh details. A promoted or + // (re)created entry embeds this post-write record rather than the fresh + // Unconfirmed details, so a confirmation wallet sync already recorded keeps + // driving graduation. let recorded = self.payment_store.get(&id).unwrap_or(details); match existing { - // The inserted entry embeds the post-write record rather than the fresh - // details, so a confirmation wallet sync already recorded keeps driving - // graduation. - None if recorded.status == PaymentStatus::Pending => { - Some(PendingPaymentDetails::new(recorded, Vec::new(), candidates)) + // First time we record this funding payment — or a crash between the two + // store writes left a Pending record with no index entry: (re)create it so + // the payment can graduate and its candidate txids stay mapped. A graduated + // payment is never `Pending`, so absence with an advanced record means the + // graduation path removed the entry and it must not be re-indexed. + None => (recorded.status == PaymentStatus::Pending).then(|| { + PendingPaymentDetails::tracked(recorded, Vec::new(), candidates, None) + }), + // A user-initiated splice has a pre-broadcast `PendingSplice` intent under + // this id; carry its intent into the `Tracked` record so the retrier can + // still clear it once the splice locks. If the payment already advanced + // beyond `Pending` (wallet sync confirmed it through `ANTI_REORG_DELAY` + // first), it must not enter the pending store; the intent stays for + // `ChannelReady` or `reconcile` to clear. + Some(PendingPaymentDetails::PendingSplice { intent, .. }) => { + if recorded.status == PaymentStatus::Pending { + Some(PendingPaymentDetails::tracked( + recorded, + Vec::new(), + candidates, + Some(intent.clone()), + )) + } else { + None + } }, - // The payment already advanced beyond Pending: the graduation path removed - // the entry and it must not be re-created. - None => None, - // The entry predates this classification — wallet sync recorded the - // transaction before it was classified (its arms and this write pair - // serialize on the cross-store lock, so nothing lands in between): merge - // only the classification into the existing entry. - Some(entry) => { + // An earlier candidate's classification or wallet sync recorded this payment + // before this classification ran (sync's arms and this write pair serialize + // on the cross-store lock, so nothing lands in between): merge only the + // classification (`tx_type`, candidate history and the figures of whichever + // candidate the record's state makes authoritative) into it. + Some(tracked @ PendingPaymentDetails::Tracked { .. }) => { + let mut updated = tracked.clone(); let pending_update = PendingPaymentDetailsUpdate { id, payment_update: Some(update), conflicting_txids: None, candidates, + splice_intent: None, }; - let mut updated = entry.clone(); updated.update(pending_update).then_some(updated) }, } @@ -1894,10 +1947,52 @@ impl Wallet { PaymentDetails::new(payment_id, kind, amount_msat, fee_paid_msat, direction, payment_status) } - fn create_pending_payment_from_tx( + /// Inserts or refreshes the pending-store entry tracking `payment` toward graduation, + /// atomically with reading the entry's current state. + async fn upsert_pending_payment( &self, payment: PaymentDetails, conflicting_txids: Vec, - ) -> PendingPaymentDetails { - PendingPaymentDetails::new(payment, conflicting_txids, Vec::new()) + ) -> Result<(), Error> { + let id = payment.id; + self.pending_payment_store + .mutate(&id, |existing| { + // Only `Pending` payments belong in the pending store. Like in + // [`Self::persist_funding_payment`], the authoritative status is re-read inside + // the store's critical section, where it cannot go stale against graduation. + let is_pending = self + .payment_store + .get(&id) + .map_or(payment.status == PaymentStatus::Pending, |recorded| { + recorded.status == PaymentStatus::Pending + }); + if !is_pending { + return None; + } + match existing { + None => { + Some(PendingPaymentDetails::new(payment, conflicting_txids, Vec::new())) + }, + // Promote a pre-broadcast splice intent: wallet sync saw the splice + // transaction before its broadcast-time classification recorded it. Carrying + // the intent into the `Tracked` record makes the entry visible to txid + // lookups while the retrier keeps the intent until the splice locks. + Some(PendingPaymentDetails::PendingSplice { intent, .. }) => { + Some(PendingPaymentDetails::tracked( + payment, + conflicting_txids, + Vec::new(), + Some(intent.clone()), + )) + }, + Some(tracked @ PendingPaymentDetails::Tracked { .. }) => { + let mut updated = tracked.clone(); + let fresh = + PendingPaymentDetails::new(payment, conflicting_txids, Vec::new()); + updated.update(fresh.to_update()).then_some(updated) + }, + } + }) + .await?; + Ok(()) } fn find_payment_by_txid(&self, target_txid: Txid) -> Option { @@ -1909,21 +2004,53 @@ impl Wallet { if let Some(replaced_details) = self .pending_payment_store .list_filter(|p| { - matches!(p.details.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid) - || p.conflicting_txids.contains(&target_txid) + p.details().is_some_and( + |d| matches!(d.kind, PaymentKind::Onchain { txid, .. } if txid == target_txid), + ) || p.conflicting_txids().contains(&target_txid) // A middle RBF round is not the record's current txid and may never have - // received a `TxReplaced` event of its own, so map any of its candidate - // txids (an earlier RBF round may confirm) back to the record. + // received a `TxReplaced` event of its own, and a splice keyed by a generated + // PaymentId is not found by the txid-derived id above: map any of the + // candidate txids (an earlier RBF round may confirm) back to the record. || p.candidate(target_txid).is_some() }) .first() { - return Some(replaced_details.details.id); + return Some(replaced_details.id()); } None } + /// Like [`Self::find_payment_by_txid`], but additionally recognizes the transaction of a + /// user-initiated splice that has no payment record yet: it spends the funding outpoint a + /// live splice intent was created for. Wallet sync can see the transaction before the + /// broadcast-time classification records it — the counterparty broadcasts it too — and must + /// adopt the splice-time `PaymentId` so both writers converge on one record. + fn find_payment_for_tx(&self, tx: &Transaction, txid: Txid) -> Option { + self.find_payment_by_txid(txid).or_else(|| { + self.pending_payment_store + .list_filter(|p| { + p.splice_intent().is_some_and(|intent| { + // A cooperative or force close also spends the funding outpoint, and a + // cooperative close pays a wallet address. A splice-in always adds wallet + // inputs while a close never has more than the funding input, so a second + // input disambiguates `SpliceKind::In`. A splice-out or fee bump can share + // the close's single-input shape; misattributing a close that races a + // still-live intent only affects which id keys its record. + let input_shape_matches = match intent.kind { + SpliceKind::In { .. } => tx.input.len() > 1, + SpliceKind::Out { .. } | SpliceKind::Rbf {} => true, + }; + let funding_txo = intent.pre_splice_funding_txo.into_bitcoin_outpoint(); + input_shape_matches + && tx.input.iter().any(|input| input.previous_output == funding_txo) + }) + }) + .first() + .map(|p| p.id()) + }) + } + /// If `payment_id` refers to a classified funding payment, refreshes its confirmation status /// and the candidate txid the event refers to, while preserving the contribution-derived /// amount/fee and `tx_type` that wallet sync must not recompute from its own view: the wallet's @@ -1992,8 +2119,7 @@ impl Wallet { // the same dual-write the default `TxConfirmed` path performs; an empty conflicting-txids // list leaves any stored conflicts intact (the update treats absent as "unchanged"). if payment.status == PaymentStatus::Pending { - let pending = self.create_pending_payment_from_tx(payment, Vec::new()); - self.pending_payment_store.insert_or_update(pending).await?; + self.upsert_pending_payment(payment, Vec::new()).await?; } Ok(true) } @@ -2236,8 +2362,6 @@ impl Wallet { ConfirmationStatus::Unconfirmed, ); - let pending_payment_store = - self.create_pending_payment_from_tx(new_payment.clone(), Vec::new()); let change_set = locked_wallet.take_staged().unwrap_or_default(); drop(locked_wallet); locked_persister.persist_changeset(change_set).await.map_err(|e| { @@ -2245,8 +2369,8 @@ impl Wallet { Error::PersistenceFailed })?; - self.payment_store.insert_or_update(new_payment).await?; - self.pending_payment_store.insert_or_update(pending_payment_store).await?; + self.payment_store.insert_or_update(new_payment.clone()).await?; + self.upsert_pending_payment(new_payment, Vec::new()).await?; self.broadcaster.broadcast_unclassified_transaction(fee_bumped_tx); @@ -4071,6 +4195,117 @@ mod tests { assert_unchanged(true); } + /// A user-initiated splice's record is keyed by the PaymentId chosen at splice time, not by + /// its funding txid. The generic funding path must resolve a rebroadcast of that funding tx + /// back to the existing record rather than minting a duplicate under the txid-derived id. + #[tokio::test] + async fn classify_funding_resolves_the_splice_time_payment_id() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let script_pubkey = wallet + .inner + .lock() + .unwrap() + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(); + let tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: Vec::new(), + output: vec![TxOut { value: Amount::from_sat(10_000), script_pubkey }], + }; + let txid = tx.compute_txid(); + + let payment_id = PaymentId([21u8; 32]); + let candidates = vec![FundingTxCandidate { + txid, + amount_msat: Some(1_000_000), + fee_paid_msat: Some(500), + }]; + let details = interactive_funding_details(payment_id, txid, Some(1_000_000), Some(500)); + wallet.persist_funding_payment(details, candidates).await.unwrap(); + + let counterparty_node_id = PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(); + let channels = vec![(counterparty_node_id, ChannelId([7u8; 32]))]; + let tx_type = TransactionType::Funding { channels: vec![] }; + wallet.classify_funding(&tx, &channels, tx_type).await.unwrap(); + + let payments = wallet.payment_store.list_filter(|_| true); + assert_eq!(payments.len(), 1, "the rebroadcast must not mint a second record"); + assert_eq!(payments[0].id, payment_id); + assert_eq!(payments[0].amount_msat, Some(1_000_000)); + assert_eq!(payments[0].fee_paid_msat, Some(500)); + } + + /// A crash between persisting a splice intent and classifying its broadcast leaves the intent + /// as the only trace of the splice-time PaymentId. The generic funding path must still + /// resolve the funding transaction — it spends the outpoint the intent was created for — + /// rather than keying the record by its txid, which the splice's own classification would + /// then duplicate. + #[tokio::test] + async fn classify_funding_resolves_a_pre_broadcast_splice_intent() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let wallet = new_test_wallet(store, false).await; + + let counterparty_node_id = PublicKey::from_str( + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + ) + .unwrap(); + let channel_id = ChannelId([7u8; 32]); + let pre_splice_funding_txo = lightning::chain::transaction::OutPoint { + txid: Txid::from_byte_array([3u8; 32]), + index: 0, + }; + + let payment_id = PaymentId([21u8; 32]); + let intent = crate::payment::pending_payment_store::SpliceIntent { + user_channel_id: crate::types::UserChannelId(42), + counterparty_node_id, + channel_id, + pre_splice_funding_txo, + contribution: crate::payment::pending_payment_store::test_funding_contribution(), + kind: SpliceKind::Rbf {}, + attempts: 0, + }; + wallet + .pending_payment_store + .insert(PendingPaymentDetails::pending_splice(payment_id, intent)) + .await + .unwrap(); + + let script_pubkey = wallet + .inner + .lock() + .unwrap() + .reveal_next_address(KeychainKind::External) + .address + .script_pubkey(); + let tx = Transaction { + version: bitcoin::transaction::Version::TWO, + lock_time: LockTime::ZERO, + input: vec![bitcoin::TxIn { + previous_output: pre_splice_funding_txo.into_bitcoin_outpoint(), + script_sig: bitcoin::ScriptBuf::new(), + sequence: bitcoin::Sequence::MAX, + witness: bitcoin::Witness::new(), + }], + output: vec![TxOut { value: Amount::from_sat(10_000), script_pubkey }], + }; + + let channels = vec![(counterparty_node_id, channel_id)]; + let tx_type = TransactionType::Funding { channels: vec![] }; + wallet.classify_funding(&tx, &channels, tx_type).await.unwrap(); + + let payments = wallet.payment_store.list_filter(|_| true); + assert_eq!(payments.len(), 1); + assert_eq!(payments[0].id, payment_id, "the record must adopt the splice-time PaymentId"); + } + /// Barrier test, classification-first ordering: wallet sync's confirmation handling must /// wait for classification's two-store write pair. Classification is parked between its /// payment-store and pending-store writes (the torn window) and only then is the diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index c1e973091..be3fa76f3 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -439,6 +439,18 @@ async fn address_pool_is_reloaded_on_restart() { expect_channel_ready_event!(node_b, node_a.node_id()); } +/// Finds an on-chain funding payment by its active candidate `txid`. A user-initiated splice's +/// `PaymentId` is generated at splice time rather than derived from a txid, so the payment must be +/// located by `kind.txid` (the active or confirmed candidate) instead of a txid-derived id. +fn funding_payment(node: &Node, txid: Txid) -> PaymentDetails { + node.list_payments_with_filter( + |p| matches!(p.kind, PaymentKind::Onchain { txid: candidate, .. } if candidate == txid), + ) + .into_iter() + .next() + .expect("no funding payment for the given txid") +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn channel_full_cycle() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); @@ -2064,9 +2076,7 @@ async fn splice_channel() { // them to the channel balance since there may not be a change output. let expected_splice_in_lightning_balance_sat = 4_000_002; - let payments = node_b.list_payments(); - let payment = - payments.into_iter().find(|p| p.id == PaymentId(txo.txid.to_byte_array())).unwrap(); + let payment = funding_payment(&node_b, txo.txid); assert_eq!(payment.fee_paid_msat, Some(expected_splice_in_fee_sat * 1_000)); assert_eq!( @@ -2117,9 +2127,7 @@ async fn splice_channel() { let expected_splice_out_fee_sat = 183; - let payments = node_a.list_payments(); - let payment = - payments.into_iter().find(|p| p.id == PaymentId(txo.txid.to_byte_array())).unwrap(); + let payment = funding_payment(&node_a, txo.txid); assert_eq!(payment.fee_paid_msat, Some(expected_splice_out_fee_sat * 1_000)); // The splice-out graduated to a confirmed interactive-funding payment. Its `direction` is left // unasserted on purpose: the destination is our own address, so it is a self-transfer (channel @@ -2144,6 +2152,129 @@ async fn splice_channel() { ); } +/// A splice on a 0conf channel locks before its funding transaction confirms: LDK promotes the +/// new funding immediately and then re-broadcasts the still-unconfirmed transaction — re-typed as +/// a generic funding transaction with wallet-view figures and no contribution data — on every +/// monitor-update completion until it confirms. The re-broadcasts must neither disturb the +/// contribution-derived record on the contributing side nor mint spurious records on either side. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn zero_conf_splice_survives_funding_rebroadcasts() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Node A's log collector synchronizes the record checks below with the re-broadcasts. + // `setup_two_nodes` wires file loggers, so build the pair manually with a collector, Node B + // trusting Node A for 0conf so channels and splices lock without confirmations. + let logger_a = Arc::new(CollectingLogWriter::new()); + let mut config_a = random_config(); + config_a.log_writer = TestLogWriter::Custom(logger_a.clone()); + let node_a = setup_node(&chain_source, config_a); + + let mut config_b = random_config(); + config_b.node_config.trusted_peers_0conf.push(node_a.node_id()); + let node_b = setup_node(&chain_source, config_b); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a], + Amount::from_sat(premine_amount_sat), + ) + .await; + node_a.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 2_000_000, false, &electrsd).await; + + // 0conf: the channel is ready without any confirmations. + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Confirm the original funding so the splice below is the only unconfirmed funding and Node + // A's change from the open is spendable for the splice contribution. + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 1_000_000).unwrap(); + let txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + wait_for_classified_funding_payment(&node_a, txo.txid).await; + + // The 0conf splice locks without confirmations, re-signaled as `ChannelReady`. + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + let payment = funding_payment(&node_a, txo.txid); + let recorded_amount_msat = payment.amount_msat; + let recorded_fee_paid_msat = payment.fee_paid_msat; + assert!(matches!( + payment.kind, + PaymentKind::Onchain { tx_type: Some(TransactionType::InteractiveFunding { .. }), .. } + )); + + // Locking the splice completed monitor updates that re-offered the unconfirmed funding + // transaction; a payment drives further monitor updates and thus further re-broadcasts. + let amount_msat = 1_000_000; + let payment_id = + node_a.spontaneous_payment().send(amount_msat, node_b.node_id(), None).unwrap(); + expect_payment_successful_event!(node_a, payment_id, None); + expect_payment_received_event!(node_b, amount_msat); + + // Wait until the classification pipeline has demonstrably processed a re-offer against the + // interactive-funding record. The broadcast loop classifies sequentially, so by the second + // arrival the first re-offer's store write has completed and the checks below are + // deterministic rather than racing the queue. + let rebroadcast = format!("funding-typed rebroadcast {}", txo.txid); + assert!( + logger_a.wait_for_count(&rebroadcast, 2).await, + "no funding re-broadcast reached Node A's classification" + ); + + let splice_payments = |node: &Node| { + node.list_payments_with_filter( + |p| matches!(p.kind, PaymentKind::Onchain { txid, .. } if txid == txo.txid), + ) + }; + + // The record must keep the splice-time id, classification, and contribution-derived figures + // through the re-broadcasts. + let payments = splice_payments(&node_a); + assert_eq!(payments.len(), 1); + let payment = &payments[0]; + assert_eq!(payment.amount_msat, recorded_amount_msat); + assert_eq!(payment.fee_paid_msat, recorded_fee_paid_msat); + assert!(matches!( + payment.kind, + PaymentKind::Onchain { tx_type: Some(TransactionType::InteractiveFunding { .. }), .. } + )); + + // Node B contributed nothing and its wallet sees no activity in the splice; the + // re-broadcasts must not mint a spurious zero-amount record for it. + assert!(splice_payments(&node_b).is_empty()); + + // Confirmation and graduation must land on that same record. + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let payments = splice_payments(&node_a); + assert_eq!(payments.len(), 1); + let payment = &payments[0]; + assert_eq!(payment.status, PaymentStatus::Succeeded); + assert_eq!(payment.amount_msat, recorded_amount_msat); + assert_eq!(payment.fee_paid_msat, recorded_fee_paid_msat); + assert!(matches!( + payment.kind, + PaymentKind::Onchain { + status: ConfirmationStatus::Confirmed { .. }, + tx_type: Some(TransactionType::InteractiveFunding { .. }), + .. + } + )); + assert!(splice_payments(&node_b).is_empty()); +} + /// Canary for the upstream behavior the zero-activity skip in `classify_funding` works around: /// after a 0conf splice is promoted, LDK re-broadcasts the still-unconfirmed funding transaction /// through its generic funding path — re-typed as a plain funding transaction without its @@ -2420,8 +2551,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // For `confirm_original`, capture the original candidate's fee and raw transaction now, before // the RBF replaces it, so it can be force-confirmed (instead of the RBF) further below. let original_candidate: Option<(Option, String)> = if confirm_original { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let fee = node_b.payment(&payment_id).expect("splice payment exists").fee_paid_msat; + let fee = funding_payment(&node_b, original_txo.txid).fee_paid_msat; let raw_tx: String = bitcoind .client .call("getrawtransaction", &[json!(original_txo.txid.to_string())]) @@ -2458,8 +2588,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // at the latest (RBF) candidate, and the durable interactive-funding `tx_type` preserved across // the replacement. let rbf_candidate_fee = { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment exists"); + let payment = funding_payment(&node_b, rbf_txo.txid); match payment.kind { PaymentKind::Onchain { txid, @@ -2473,7 +2602,9 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { }, } assert_eq!(payment.status, PaymentStatus::Pending); - // Only one Onchain Pending payment for this splice attempt (not one per candidate). + // Only one Onchain Pending payment for this splice attempt (not one per candidate). This also + // guards the intent-clobber fix: had the sync above cleared this splice's live intent, the + // bump would not have found it and would have minted a second record under a fresh PaymentId. let splice_payments = node_b.list_payments_with_filter(|p| { p.direction == PaymentDirection::Outbound && matches!(p.kind, PaymentKind::Onchain { .. }) @@ -2533,8 +2664,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { // channel-lifecycle signal, not what drives payment status. Its `kind.txid` reflects the // winning RBF candidate, and `fee_paid_msat` carries this node's `FundingContribution` fee. { - let payment_id = PaymentId(original_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment graduated"); + let payment = funding_payment(&node_b, winning_txo.txid); assert_eq!(payment.status, PaymentStatus::Succeeded); match payment.kind { PaymentKind::Onchain { txid, status: ConfirmationStatus::Confirmed { .. }, .. } => { @@ -2657,8 +2787,7 @@ async fn splice_payment_reorged_to_unconfirmed() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; node_b.sync_wallets().unwrap(); - let payment_id = PaymentId(splice_txo.txid.to_byte_array()); - let payment = node_b.payment(&payment_id).expect("splice payment exists"); + let payment = funding_payment(&node_b, splice_txo.txid); assert_eq!(payment.status, PaymentStatus::Pending); assert!(matches!( payment.kind, @@ -2681,7 +2810,7 @@ async fn splice_payment_reorged_to_unconfirmed() { // The funding payment returns to `Unconfirmed` and stays `Pending`, exercising the // `TxUnconfirmed` arm for a funding payment. - let payment = node_b.payment(&payment_id).expect("splice payment still exists"); + let payment = funding_payment(&node_b, splice_txo.txid); assert_eq!(payment.status, PaymentStatus::Pending); assert!(matches!( payment.kind, @@ -2692,6 +2821,301 @@ async fn splice_payment_reorged_to_unconfirmed() { node_b.stop().unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_resumed_after_restart() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let onchain_balance_before_sat = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Initiate a splice-out while disconnected: LDK accepts the contribution but cannot make + // progress before the restart below drops it, having neither negotiated nor persisted + // anything. Only the persisted splice intent allows resuming the splice. + node_a.disconnect(node_b.node_id()).unwrap(); + let address = node_a.onchain_payment().new_address().unwrap(); + node_a.splice_out(&user_channel_id_a, node_b.node_id(), &address, 500_000).unwrap(); + + let onchain_balance_before_sat = node_a.list_balances().total_onchain_balance_sats; + node_a.stop().unwrap(); + onchain_balance_before_sat + }; + + // On restart, the reconciler resubmits the splice, which proceeds once the peer connects. + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + + let txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + + wait_for_tx(&electrsd.client, txo.txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + assert!( + node_a.list_balances().total_onchain_balance_sats > onchain_balance_before_sat + 400_000, + "resumed splice-out should have moved ~500k sats to the on-chain balance", + ); + + // The locked splice cleared the intent, so another restart must not resubmit it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "completed splice should not be resubmitted"); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_rbf_resumed_after_restart() { + // Use a custom bitcoind config with a lower incrementalrelayfee so that the +25 sat/kwu + // (0.1 sat/vB) RBF feerate bump satisfies BIP125's absolute fee increase requirement. + let bitcoind_exe = std::env::var("BITCOIND_EXE") + .ok() + .or_else(|| corepc_node::downloaded_exe_path().ok()) + .expect( + "you need to provide an env var BITCOIND_EXE or specify a bitcoind version feature", + ); + let mut bitcoind_conf = corepc_node::Conf::default(); + bitcoind_conf.network = "regtest"; + bitcoind_conf.args.push("-rest"); + bitcoind_conf.args.push("-incrementalrelayfee=0.00000100"); + let bitcoind = BitcoinD::with_conf(bitcoind_exe, &bitcoind_conf).unwrap(); + + let electrs_exe = std::env::var("ELECTRS_EXE") + .ok() + .or_else(electrsd::downloaded_exe_path) + .expect("you need to provide env var ELECTRS_EXE or specify an electrsd version feature"); + let mut electrsd_conf = electrsd::Conf::default(); + electrsd_conf.http_enabled = true; + electrsd_conf.network = "regtest"; + let electrsd = ElectrsD::with_conf(electrs_exe, &bitcoind, &electrsd_conf).unwrap(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let original_txo = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // Negotiate a splice but leave its transaction unconfirmed so it can be fee-bumped. + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 500_000).unwrap(); + let original_txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + wait_for_tx(&electrsd.client, original_txo.txid).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + // Bump the fee while disconnected and restart before anything could be negotiated: only + // the persisted intent knows about the fee bump, while LDK still has the negotiated + // splice at the original feerate. + node_a.disconnect(node_b.node_id()).unwrap(); + node_a.bump_channel_funding_fee(&user_channel_id_a, node_b.node_id()).unwrap(); + node_a.stop().unwrap(); + original_txo + }; + + // On restart, the reconciler sees that the negotiated splice is still at a lower feerate + // than the persisted fee-bump intent and resubmits the bump. + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + + let rbf_txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + assert_ne!(original_txo, rbf_txo, "resubmitted RBF should produce a different funding txo"); + + // Restarting again must not resubmit the bump: the negotiated splice now carries it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a.clone()); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr.clone(), false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "carried-out fee bump should not be resubmitted"); + + wait_for_tx(&electrsd.client, rbf_txo.txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + // The locked fee bump cleared its intent, so a further restart must not resubmit it. + node_a.stop().unwrap(); + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + assert!(node_a.next_event().is_none(), "locked fee bump should not be resubmitted"); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn splice_payment_tracked_across_restart_before_lock() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + // Set up node_a manually so it can be restarted with the same config. + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + let splice_txid = { + let node_a = setup_node(&chain_source, config_a.clone()); + + let address_a = node_a.onchain_payment().new_address().unwrap(); + let address_b = node_b.onchain_payment().new_address().unwrap(); + let premine_amount_sat = 5_000_000; + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![address_a, address_b], + Amount::from_sat(premine_amount_sat), + ) + .await; + + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + node_a.splice_in(&user_channel_id_a, node_b.node_id(), 500_000).unwrap(); + let txo = expect_splice_negotiated_event!(node_a, node_b.node_id()); + + // Stop node_a as soon as the splice is negotiated. node_b broadcasts the transaction + // either way, so it reaches the chain while node_a is offline. Depending on timing, + // node_a may or may not have classified its own broadcast into a payment record before + // stopping; the assertions below must hold in both cases. + node_a.stop().unwrap(); + txo.txid + }; + + // Confirm the splice while node_a is offline, but keep it short of the depth at which it + // locks, so node_a restarts with its splice intent still live. + wait_for_tx(&electrsd.client, splice_txid).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; + + // After the restart, wallet sync and classification must agree on the splice-time + // `PaymentId` no matter which of them sees the confirmed transaction first: exactly one + // payment record, and not one keyed by a txid-derived id. + let node_a = setup_node(&chain_source, config_a); + node_a.sync_wallets().unwrap(); + + let splice_payments = |node: &Node| { + node.list_payments_with_filter( + |p| matches!(p.kind, PaymentKind::Onchain { txid, .. } if txid == splice_txid), + ) + }; + let payments = splice_payments(&node_a); + assert_eq!( + payments.len(), + 1, + "expected exactly one payment record for the splice, got {}: {:#?}", + payments.len(), + payments, + ); + assert_ne!( + payments[0].id, + PaymentId(splice_txid.to_byte_array()), + "the splice payment must keep its splice-time id, not a txid-derived fallback", + ); + assert_eq!(payments[0].status, PaymentStatus::Pending); + + // Reconnect and let the splice lock: the single record graduates instead of gaining a + // duplicate. + let node_b_addr = node_b.listening_addresses().unwrap().first().unwrap().clone(); + node_a.connect(node_b.node_id(), node_b_addr, false).unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 5).await; + node_a.sync_wallets().unwrap(); + node_b.sync_wallets().unwrap(); + + expect_channel_ready_event!(node_a, node_b.node_id()); + expect_channel_ready_event!(node_b, node_a.node_id()); + + let payments = splice_payments(&node_a); + assert_eq!( + payments.len(), + 1, + "expected exactly one payment record after the splice locked, got {}: {:#?}", + payments.len(), + payments, + ); + assert_eq!(payments[0].status, PaymentStatus::Succeeded); + + node_a.stop().unwrap(); + node_b.stop().unwrap(); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn splice_in_rbf_joins_counterparty_splice() { let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();