From 782e8c0ec9980a2b2c7d42080163328e8d86720c Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Tue, 17 Mar 2026 12:12:32 -0400 Subject: [PATCH 01/20] Split up some tests that have many variants Helps when debugging to know which variants failed. --- lightning/src/ln/monitor_tests.rs | 60 +++++++++++++++++++++++++++++-- lightning/src/ln/payment_tests.rs | 26 +++++++++++++- 2 files changed, 83 insertions(+), 3 deletions(-) diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs index f52f093917b..d156f874703 100644 --- a/lightning/src/ln/monitor_tests.rs +++ b/lightning/src/ln/monitor_tests.rs @@ -3803,27 +3803,83 @@ fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: b } #[test] -fn test_lost_timeout_monitor_events() { +fn test_lost_timeout_monitor_events_a() { do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, false, false); +} +#[test] +fn test_lost_timeout_monitor_events_b() { do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, true, false); +} +#[test] +fn test_lost_timeout_monitor_events_c() { do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, false, false); +} +#[test] +fn test_lost_timeout_monitor_events_d() { do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, true, false); +} +#[test] +fn test_lost_timeout_monitor_events_e() { do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, false, false); +} +#[test] +fn test_lost_timeout_monitor_events_f() { do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, true, false); +} +#[test] +fn test_lost_timeout_monitor_events_g() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, false, false); +} +#[test] +fn test_lost_timeout_monitor_events_h() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, true, false); +} +#[test] +fn test_lost_timeout_monitor_events_i() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, false, false); +} +#[test] +fn test_lost_timeout_monitor_events_j() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, true, false); - +} +#[test] +fn test_lost_timeout_monitor_events_k() { do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, false, true); +} +#[test] +fn test_lost_timeout_monitor_events_l() { do_test_lost_timeout_monitor_events(CommitmentType::RevokedCounterparty, true, true); +} +#[test] +fn test_lost_timeout_monitor_events_m() { do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, false, true); +} +#[test] +fn test_lost_timeout_monitor_events_n() { do_test_lost_timeout_monitor_events(CommitmentType::PreviousCounterparty, true, true); +} +#[test] +fn test_lost_timeout_monitor_events_o() { do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, false, true); +} +#[test] +fn test_lost_timeout_monitor_events_p() { do_test_lost_timeout_monitor_events(CommitmentType::LatestCounterparty, true, true); +} +#[test] +fn test_lost_timeout_monitor_events_q() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, false, true); +} +#[test] +fn test_lost_timeout_monitor_events_r() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithoutLastHTLC, true, true); +} +#[test] +fn test_lost_timeout_monitor_events_s() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, false, true); +} +#[test] +fn test_lost_timeout_monitor_events_t() { do_test_lost_timeout_monitor_events(CommitmentType::LocalWithLastHTLC, true, true); } diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs index 2eb5d4ee85c..445fa1a1eea 100644 --- a/lightning/src/ln/payment_tests.rs +++ b/lightning/src/ln/payment_tests.rs @@ -1465,15 +1465,39 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload( } #[test] -fn test_dup_htlc_onchain_doesnt_fail_on_reload() { +fn test_dup_htlc_onchain_doesnt_fail_on_reload_a() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, true, true); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_b() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, true, false); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_c() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, true, false, false); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_d() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, true, true); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_e() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, true, false); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_f() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(true, false, false, false); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_g() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, true, true); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_h() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, true, false); +} +#[test] +fn test_dup_htlc_onchain_doesnt_fail_on_reload_i() { do_test_dup_htlc_onchain_doesnt_fail_on_reload(false, false, false, false); } From 24c1d2446b6b76955d738013472adb5f027241b2 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Thu, 19 Mar 2026 16:08:46 -0400 Subject: [PATCH 02/20] Remove unnecessary pending_monitor_events clone --- lightning/src/chain/channelmonitor.rs | 31 +++++++++++++-------------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 42d04e0f8ce..27d2fed81dc 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -69,8 +69,8 @@ use crate::util::byte_utils; use crate::util::logger::{Logger, WithContext}; use crate::util::persist::MonitorName; use crate::util::ser::{ - MaybeReadable, Readable, ReadableArgs, RequiredWrapper, UpgradableRequired, Writeable, Writer, - U48, + Iterable, MaybeReadable, Readable, ReadableArgs, RequiredWrapper, UpgradableRequired, + Writeable, Writer, U48, }; #[allow(unused_imports)] @@ -1734,24 +1734,23 @@ pub(crate) fn write_chanmon_internal( channel_monitor.lockdown_from_offchain.write(writer)?; channel_monitor.holder_tx_signed.write(writer)?; - // If we have a `HolderForceClosedWithInfo` event, we need to write the `HolderForceClosed` for backwards compatibility. - let pending_monitor_events = - match channel_monitor.pending_monitor_events.iter().find(|ev| match ev { - MonitorEvent::HolderForceClosedWithInfo { .. } => true, - _ => false, - }) { - Some(MonitorEvent::HolderForceClosedWithInfo { outpoint, .. }) => { - let mut pending_monitor_events = channel_monitor.pending_monitor_events.clone(); - pending_monitor_events.push(MonitorEvent::HolderForceClosed(*outpoint)); - pending_monitor_events - }, - _ => channel_monitor.pending_monitor_events.clone(), - }; + // If we have a `HolderForceClosedWithInfo` event, we need to write the `HolderForceClosed` + // for backwards compatibility. + let holder_force_closed_compat = channel_monitor.pending_monitor_events.iter().find_map(|ev| { + if let MonitorEvent::HolderForceClosedWithInfo { outpoint, .. } = ev { + Some(MonitorEvent::HolderForceClosed(*outpoint)) + } else { + None + } + }); + let pending_monitor_events = Iterable( + channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()), + ); write_tlv_fields!(writer, { (1, channel_monitor.funding_spend_confirmed, option), (3, channel_monitor.htlcs_resolved_on_chain, required_vec), - (5, pending_monitor_events, required_vec), + (5, pending_monitor_events, required), // Equivalent to required_vec because Iterable also writes as WithoutLength (7, channel_monitor.funding_spend_seen, required), (9, channel_monitor.counterparty_node_id, required), (11, channel_monitor.confirmed_commitment_tx_counterparty_output, option), From 63d9ca3cf8d48ecf85fd02f87bde2d2140ae8f7b Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Tue, 17 Mar 2026 15:26:43 -0400 Subject: [PATCH 03/20] Add persistent_monitor_events flag to monitors/manager Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. As a first step towards this, here we persist a flag in the ChannelManager and ChannelMonitors indicating whether this new feature is enabled. It will be used in upcoming commits to maintain compatibility and create an upgrade/downgrade path between LDK versions. --- lightning/src/chain/channelmonitor.rs | 20 +++++++++++++++++++ lightning/src/ln/channelmanager.rs | 28 +++++++++++++++++++++++++++ 2 files changed, 48 insertions(+) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 27d2fed81dc..d41970416f3 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -1296,6 +1296,12 @@ pub(crate) struct ChannelMonitorImpl { // block/transaction-connected events and *not* during block/transaction-disconnected events, // we further MUST NOT generate events during block/transaction-disconnection. pending_monitor_events: Vec, + /// When set, monitor events are retained until explicitly acked rather than cleared on read. + /// + /// Allows the ChannelManager to reconstruct pending HTLC state by replaying monitor events on + /// startup, and make the monitor responsible for both off- and on-chain payment resolution. Will + /// be always set once support for this feature is complete. + persistent_events_enabled: bool, pub(super) pending_events: Vec, pub(super) is_processing_pending_events: bool, @@ -1747,8 +1753,12 @@ pub(crate) fn write_chanmon_internal( channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()), ); + // Only write `persistent_events_enabled` if it's set to true, as it's an even TLV. + let persistent_events_enabled = channel_monitor.persistent_events_enabled.then_some(()); + write_tlv_fields!(writer, { (1, channel_monitor.funding_spend_confirmed, option), + (2, persistent_events_enabled, option), (3, channel_monitor.htlcs_resolved_on_chain, required_vec), (5, pending_monitor_events, required), // Equivalent to required_vec because Iterable also writes as WithoutLength (7, channel_monitor.funding_spend_seen, required), @@ -1956,6 +1966,7 @@ impl ChannelMonitor { payment_preimages: new_hash_map(), pending_monitor_events: Vec::new(), + persistent_events_enabled: false, pending_events: Vec::new(), is_processing_pending_events: false, @@ -6742,8 +6753,10 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP let mut funding_seen_onchain = RequiredWrapper(None); let mut best_block_previous_blocks = None; let mut current_funding_contribution = None; + let mut persistent_events_enabled: Option<()> = None; read_tlv_fields!(reader, { (1, funding_spend_confirmed, option), + (2, persistent_events_enabled, option), (3, htlcs_resolved_on_chain, optional_vec), (5, pending_monitor_events, optional_vec), (7, funding_spend_seen, option), @@ -6771,6 +6784,12 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP best_block.previous_blocks = previous_blocks; } + #[cfg(not(any(feature = "_test_utils", test)))] + if persistent_events_enabled.is_some() { + // This feature isn't supported yet so error if the writer expected it to be. + return Err(DecodeError::InvalidValue) + } + // Note that `payment_preimages_with_info` was added (and is always written) in LDK 0.1, so // we can use it to determine if this monitor was last written by LDK 0.1 or later. let written_by_0_1_or_later = payment_preimages_with_info.is_some(); @@ -6920,6 +6939,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP payment_preimages, pending_monitor_events: pending_monitor_events.unwrap(), + persistent_events_enabled: persistent_events_enabled.is_some(), pending_events, is_processing_pending_events: false, diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 25c86b3f0cb..a40c0204acf 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -3011,6 +3011,15 @@ pub struct ChannelManager< /// [`ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee`] estimate. last_days_feerates: Mutex>, + /// When set, monitors will repeatedly provide an event back to the `ChannelManager` on restart + /// until the event is explicitly acknowledged as processed. + /// + /// Allows us to reconstruct pending HTLC state by replaying monitor events on startup, rather + /// than from complexly polling and reconciling Channel{Monitor} APIs, as well as move the + /// responsibility of off-chain payment resolution from the Channel to the monitor. Will be + /// always set once support is complete. + persistent_monitor_events: bool, + #[cfg(test)] pub(super) entropy_source: ES, #[cfg(not(test))] @@ -3747,6 +3756,8 @@ impl< signer_provider, logger, + + persistent_monitor_events: false, } } @@ -18412,6 +18423,9 @@ impl< } } + // Only write `persistent_events_enabled` if it's set to true, as it's an even TLV. + let persistent_monitor_events = self.persistent_monitor_events.then_some(()); + write_tlv_fields!(writer, { (1, pending_outbound_payments_no_retry, required), (2, pending_intercepted_htlcs, option), @@ -18424,6 +18438,7 @@ impl< (9, htlc_purposes, required_vec), (10, legacy_in_flight_monitor_updates, option), (11, self.probing_cookie_secret, required), + (12, persistent_monitor_events, option), (13, htlc_onion_fields, optional_vec), (14, decode_update_add_htlcs_opt, option), (15, self.inbound_payment_id_secret, required), @@ -18523,6 +18538,7 @@ pub(super) struct ChannelManagerData { forward_htlcs_legacy: HashMap>, pending_intercepted_htlcs_legacy: HashMap, decode_update_add_htlcs_legacy: HashMap>, + persistent_monitor_events: bool, // The `ChannelManager` version that was written. version: u8, } @@ -18709,6 +18725,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger> let mut peer_storage_dir: Option)>> = None; let mut async_receive_offer_cache: AsyncReceiveOfferCache = AsyncReceiveOfferCache::new(); let mut best_block_previous_blocks = None; + let mut persistent_monitor_events: Option<()> = None; read_tlv_fields!(reader, { (1, pending_outbound_payments_no_retry, option), (2, pending_intercepted_htlcs_legacy, option), @@ -18721,6 +18738,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger> (9, claimable_htlc_purposes, optional_vec), (10, legacy_in_flight_monitor_updates, option), (11, probing_cookie_secret, option), + (12, persistent_monitor_events, option), (13, amountless_claimable_htlc_onion_fields, optional_vec), (14, decode_update_add_htlcs_legacy, option), (15, inbound_payment_id_secret, option), @@ -18730,6 +18748,12 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger> (23, best_block_previous_blocks, option), }); + #[cfg(not(any(feature = "_test_utils", test)))] + if persistent_monitor_events.is_some() { + // This feature isn't supported yet so error if the writer expected it to be. + return Err(DecodeError::InvalidValue); + } + // Merge legacy pending_outbound_payments fields into a single HashMap. // Priority: pending_outbound_payments (TLV 3) > pending_outbound_payments_no_retry (TLV 1) // > pending_outbound_payments_compat (non-TLV legacy) @@ -18849,6 +18873,7 @@ impl<'a, ES: EntropySource, SP: SignerProvider, L: Logger> peer_storage_dir: peer_storage_dir.unwrap_or_default(), async_receive_offer_cache, version, + persistent_monitor_events: persistent_monitor_events.is_some(), }) } } @@ -19149,6 +19174,7 @@ impl< mut in_flight_monitor_updates, peer_storage_dir, async_receive_offer_cache, + persistent_monitor_events, version: _version, } = data; @@ -20415,6 +20441,8 @@ impl< logger: args.logger, config: RwLock::new(args.config), + + persistent_monitor_events, }; let mut processed_claims: HashSet> = new_hash_set(); From 88751bd34fbb5d067549caaea32cfe30ddd72fa1 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Wed, 18 Mar 2026 15:08:54 -0400 Subject: [PATCH 04/20] Add helper to push monitor events Cleans up the next commit --- lightning/src/chain/chainmonitor.rs | 53 +++++++-------------------- lightning/src/chain/channelmonitor.rs | 26 +++++++++---- 2 files changed, 32 insertions(+), 47 deletions(-) diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index b3b69096997..6d105ff1d3f 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -366,9 +366,6 @@ pub struct ChainMonitor< fee_estimator: F, persister: P, _entropy_source: ES, - /// "User-provided" (ie persistence-completion/-failed) [`MonitorEvent`]s. These came directly - /// from the user and not from a [`ChannelMonitor`]. - pending_monitor_events: Mutex, PublicKey)>>, /// The best block height seen, used as a proxy for the passage of time. highest_chain_height: AtomicUsize, @@ -436,7 +433,6 @@ where logger, fee_estimator: feeest, _entropy_source, - pending_monitor_events: Mutex::new(Vec::new()), highest_chain_height: AtomicUsize::new(0), event_notifier: Arc::clone(&event_notifier), persister: AsyncPersister { persister, event_notifier }, @@ -657,7 +653,6 @@ where fee_estimator: feeest, persister, _entropy_source, - pending_monitor_events: Mutex::new(Vec::new()), highest_chain_height: AtomicUsize::new(0), event_notifier: Arc::new(Notifier::new()), pending_send_only_events: Mutex::new(Vec::new()), @@ -802,16 +797,11 @@ where return Ok(()); } let funding_txo = monitor_data.monitor.get_funding_txo(); - self.pending_monitor_events.lock().unwrap().push(( + monitor_data.monitor.push_monitor_event(MonitorEvent::Completed { funding_txo, channel_id, - vec![MonitorEvent::Completed { - funding_txo, - channel_id, - monitor_update_id: monitor_data.monitor.get_latest_update_id(), - }], - monitor_data.monitor.get_counterparty_node_id(), - )); + monitor_update_id: monitor_data.monitor.get_latest_update_id(), + }); self.event_notifier.notify(); Ok(()) @@ -824,14 +814,11 @@ where pub fn force_channel_monitor_updated(&self, channel_id: ChannelId, monitor_update_id: u64) { let monitors = self.monitors.read().unwrap(); let monitor = &monitors.get(&channel_id).unwrap().monitor; - let counterparty_node_id = monitor.get_counterparty_node_id(); - let funding_txo = monitor.get_funding_txo(); - self.pending_monitor_events.lock().unwrap().push(( - funding_txo, + monitor.push_monitor_event(MonitorEvent::Completed { + funding_txo: monitor.get_funding_txo(), channel_id, - vec![MonitorEvent::Completed { funding_txo, channel_id, monitor_update_id }], - counterparty_node_id, - )); + monitor_update_id, + }); self.event_notifier.notify(); } @@ -1266,21 +1253,13 @@ where // The channel is post-close (funding spend seen, lockdown, or // holder tx signed). Return InProgress so ChannelManager freezes // the channel until the force-close MonitorEvents are processed. - // Push a Completed event into pending_monitor_events so it gets - // picked up after the per-monitor events in the next - // release_pending_monitor_events call. - let funding_txo = monitor.get_funding_txo(); - let channel_id = monitor.channel_id(); - self.pending_monitor_events.lock().unwrap().push(( - funding_txo, - channel_id, - vec![MonitorEvent::Completed { - funding_txo, - channel_id, - monitor_update_id: monitor.get_latest_update_id(), - }], - monitor.get_counterparty_node_id(), - )); + // Push a Completed event into the monitor so it gets picked up + // in the next release_pending_monitor_events call. + monitor.push_monitor_event(MonitorEvent::Completed { + funding_txo: monitor.get_funding_txo(), + channel_id: monitor.channel_id(), + monitor_update_id: monitor.get_latest_update_id(), + }); log_debug!( logger, "Deferring completion of ChannelMonitorUpdate id {:?} (channel is post-close)", @@ -1665,10 +1644,6 @@ where )); } } - // Drain pending_monitor_events (which includes deferred post-close - // completions) after per-monitor events so that force-close - // MonitorEvents are processed by ChannelManager first. - pending_monitor_events.extend(self.pending_monitor_events.lock().unwrap().split_off(0)); pending_monitor_events } } diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index d41970416f3..e7cc8ad49e6 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -184,6 +184,10 @@ impl Readable for ChannelMonitorUpdate { } } +fn push_monitor_event(pending_monitor_events: &mut Vec, event: MonitorEvent) { + pending_monitor_events.push(event); +} + /// An event to be processed by the ChannelManager. #[derive(Clone, PartialEq, Eq)] pub enum MonitorEvent { @@ -227,8 +231,6 @@ pub enum MonitorEvent { }, } impl_writeable_tlv_based_enum_upgradable_legacy!(MonitorEvent, - // Note that Completed is currently never serialized to disk as it is generated only in - // ChainMonitor. (0, Completed) => { (0, funding_txo, required), (2, monitor_update_id, required), @@ -2184,6 +2186,10 @@ impl ChannelMonitor { self.inner.lock().unwrap().get_and_clear_pending_monitor_events() } + pub(super) fn push_monitor_event(&self, event: MonitorEvent) { + self.inner.lock().unwrap().push_monitor_event(event); + } + /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. /// /// For channels featuring anchor outputs, this method will also process [`BumpTransaction`] @@ -3909,7 +3915,7 @@ impl ChannelMonitorImpl { outpoint: funding_outpoint, channel_id: self.channel_id, }; - self.pending_monitor_events.push(event); + push_monitor_event(&mut self.pending_monitor_events, event); } // Although we aren't signing the transaction directly here, the transaction will be signed @@ -4599,6 +4605,10 @@ impl ChannelMonitorImpl { &self.outputs_to_watch } + fn push_monitor_event(&mut self, event: MonitorEvent) { + push_monitor_event(&mut self.pending_monitor_events, event); + } + fn get_and_clear_pending_monitor_events(&mut self) -> Vec { let mut ret = Vec::new(); mem::swap(&mut ret, &mut self.pending_monitor_events); @@ -5658,7 +5668,7 @@ impl ChannelMonitorImpl { ); log_info!(logger, "Channel closed by funding output spend in txid {txid}"); if !self.funding_spend_seen { - self.pending_monitor_events.push(MonitorEvent::CommitmentTxConfirmed(())); + self.push_monitor_event(MonitorEvent::CommitmentTxConfirmed(())); } self.funding_spend_seen = true; @@ -5833,7 +5843,7 @@ impl ChannelMonitorImpl { log_debug!(logger, "HTLC {} failure update in {} has got enough confirmations to be passed upstream", &payment_hash, entry.txid); - self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate { + self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate { payment_hash, payment_preimage: None, source, @@ -5942,7 +5952,7 @@ impl ChannelMonitorImpl { log_error!(logger, "Failing back HTLC {} upstream to preserve the \ channel as the forward HTLC hasn't resolved and our backward HTLC \ expires soon at {}", log_bytes!(htlc.payment_hash.0), inbound_htlc_expiry); - self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate { + push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate { source: source.clone(), payment_preimage: None, payment_hash: htlc.payment_hash, @@ -6359,7 +6369,7 @@ impl ChannelMonitorImpl { }, }); self.counterparty_fulfilled_htlcs.insert(SentHTLCId::from_source(&source), payment_preimage); - self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate { + push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate { source, payment_preimage: Some(payment_preimage), payment_hash, @@ -6383,7 +6393,7 @@ impl ChannelMonitorImpl { }, }); self.counterparty_fulfilled_htlcs.insert(SentHTLCId::from_source(&source), payment_preimage); - self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate { + push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate { source, payment_preimage: Some(payment_preimage), payment_hash, From 1338cb80357cb779428cee3f786c8576de5139ee Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Wed, 18 Mar 2026 17:18:26 -0400 Subject: [PATCH 05/20] Rename pending_monitor_events to _legacy This field will be deprecated in upcoming commits when we start persisting MonitorEvent ids alongside the MonitorEvents. --- lightning/src/chain/channelmonitor.rs | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index e7cc8ad49e6..7cfc5564215 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -1751,7 +1751,7 @@ pub(crate) fn write_chanmon_internal( None } }); - let pending_monitor_events = Iterable( + let pending_monitor_events_legacy = Iterable( channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()), ); @@ -1762,7 +1762,7 @@ pub(crate) fn write_chanmon_internal( (1, channel_monitor.funding_spend_confirmed, option), (2, persistent_events_enabled, option), (3, channel_monitor.htlcs_resolved_on_chain, required_vec), - (5, pending_monitor_events, required), // Equivalent to required_vec because Iterable also writes as WithoutLength + (5, pending_monitor_events_legacy, required), // Equivalent to required_vec because Iterable also writes as WithoutLength (7, channel_monitor.funding_spend_seen, required), (9, channel_monitor.counterparty_node_id, required), (11, channel_monitor.confirmed_commitment_tx_counterparty_output, option), @@ -6691,16 +6691,16 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP } } - let pending_monitor_events_len: u64 = Readable::read(reader)?; - let mut pending_monitor_events = Some( - Vec::with_capacity(cmp::min(pending_monitor_events_len as usize, MAX_ALLOC_SIZE / (32 + 8*3)))); - for _ in 0..pending_monitor_events_len { + let pending_monitor_events_legacy_len: u64 = Readable::read(reader)?; + let mut pending_monitor_events_legacy = Some( + Vec::with_capacity(cmp::min(pending_monitor_events_legacy_len as usize, MAX_ALLOC_SIZE / (32 + 8*3)))); + for _ in 0..pending_monitor_events_legacy_len { let ev = match ::read(reader)? { 0 => MonitorEvent::HTLCEvent(Readable::read(reader)?), 1 => MonitorEvent::HolderForceClosed(outpoint), _ => return Err(DecodeError::InvalidValue) }; - pending_monitor_events.as_mut().unwrap().push(ev); + pending_monitor_events_legacy.as_mut().unwrap().push(ev); } let pending_events_len: u64 = Readable::read(reader)?; @@ -6768,7 +6768,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP (1, funding_spend_confirmed, option), (2, persistent_events_enabled, option), (3, htlcs_resolved_on_chain, optional_vec), - (5, pending_monitor_events, optional_vec), + (5, pending_monitor_events_legacy, optional_vec), (7, funding_spend_seen, option), (9, counterparty_node_id, option), (11, confirmed_commitment_tx_counterparty_output, option), @@ -6822,11 +6822,11 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP // `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. If we have both // events, we can remove the `HolderForceClosed` event and just keep the `HolderForceClosedWithInfo`. - if let Some(ref mut pending_monitor_events) = pending_monitor_events { - if pending_monitor_events.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosed(_))) && - pending_monitor_events.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosedWithInfo { .. })) + if let Some(ref mut evs) = pending_monitor_events_legacy { + if evs.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosed(_))) && + evs.iter().any(|e| matches!(e, MonitorEvent::HolderForceClosedWithInfo { .. })) { - pending_monitor_events.retain(|e| !matches!(e, MonitorEvent::HolderForceClosed(_))); + evs.retain(|e| !matches!(e, MonitorEvent::HolderForceClosed(_))); } } @@ -6948,7 +6948,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP current_holder_commitment_number, payment_preimages, - pending_monitor_events: pending_monitor_events.unwrap(), + pending_monitor_events: pending_monitor_events_legacy.unwrap(), persistent_events_enabled: persistent_events_enabled.is_some(), pending_events, is_processing_pending_events: false, From b9622859b8bd39e3fd2b43216d46a59c0f8f995e Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Mon, 16 Mar 2026 13:37:38 -0400 Subject: [PATCH 06/20] Add chain::Watch ack_monitor_event API Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. Here we add an as-yet-unused API to chain::Watch to allow the ChannelManager to tell the a ChannelMonitor that a MonitorEvent has been irrevocably processed and can be deleted. --- lightning/src/chain/chainmonitor.rs | 24 ++++++++++++++++++++++++ lightning/src/chain/channelmonitor.rs | 6 ++++++ lightning/src/chain/mod.rs | 14 ++++++++++++++ lightning/src/util/test_utils.rs | 6 +++++- 4 files changed, 49 insertions(+), 1 deletion(-) diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index 6d105ff1d3f..8832d6fec71 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -66,6 +66,21 @@ use core::iter::Cycle; use core::ops::Deref; use core::sync::atomic::{AtomicUsize, Ordering}; +/// Identifies the source of a [`MonitorEvent`] for acknowledgment via +/// [`chain::Watch::ack_monitor_event`] once the event has been processed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct MonitorEventSource { + /// The event ID assigned by the [`ChannelMonitor`]. + pub event_id: u64, + /// The channel from which the [`MonitorEvent`] originated. + pub channel_id: ChannelId, +} + +impl_writeable_tlv_based!(MonitorEventSource, { + (1, event_id, required), + (3, channel_id, required), +}); + /// A pending operation queued for later execution when `ChainMonitor` is in deferred mode. enum PendingMonitorOp { /// A new monitor to insert and persist. @@ -1646,6 +1661,15 @@ where } pending_monitor_events } + + fn ack_monitor_event(&self, source: MonitorEventSource) { + let monitors = self.monitors.read().unwrap(); + if let Some(monitor_state) = monitors.get(&source.channel_id) { + monitor_state.monitor.ack_monitor_event(source.event_id); + } else { + debug_assert!(false, "Ack'd monitor events should always have a corresponding monitor"); + } + } } impl< diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 7cfc5564215..aa4ecaaa75c 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -2190,6 +2190,12 @@ impl ChannelMonitor { self.inner.lock().unwrap().push_monitor_event(event); } + /// Removes a [`MonitorEvent`] by its event ID, acknowledging that it has been processed. + /// Generally called by [`chain::Watch::ack_monitor_event`]. + pub fn ack_monitor_event(&self, _event_id: u64) { + // TODO: once events have ids, remove the corresponding event here + } + /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. /// /// For channels featuring anchor outputs, this method will also process [`BumpTransaction`] diff --git a/lightning/src/chain/mod.rs b/lightning/src/chain/mod.rs index d72d58b3149..ebd1fddaa90 100644 --- a/lightning/src/chain/mod.rs +++ b/lightning/src/chain/mod.rs @@ -18,6 +18,7 @@ use bitcoin::network::Network; use bitcoin::script::{Script, ScriptBuf}; use bitcoin::secp256k1::PublicKey; +use crate::chain::chainmonitor::MonitorEventSource; use crate::chain::channelmonitor::{ ChannelMonitor, ChannelMonitorUpdate, MonitorEvent, ANTI_REORG_DELAY, }; @@ -427,6 +428,15 @@ pub trait Watch { fn release_pending_monitor_events( &self, ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)>; + + /// Acknowledges and removes a [`MonitorEvent`] previously returned by + /// [`Watch::release_pending_monitor_events`] by its event ID. + /// + /// Once acknowledged, the event will no longer be returned by future calls to + /// [`Watch::release_pending_monitor_events`] and will not be replayed on restart. + /// + /// Events may be acknowledged in any order. + fn ack_monitor_event(&self, source: MonitorEventSource); } impl + ?Sized, W: Deref> @@ -449,6 +459,10 @@ impl + ?Sized, W: Der ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)> { self.deref().release_pending_monitor_events() } + + fn ack_monitor_event(&self, source: MonitorEventSource) { + self.deref().ack_monitor_event(source) + } } /// The `Filter` trait defines behavior for indicating chain activity of interest pertaining to diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs index 892c9f4169d..f77e9ca39d4 100644 --- a/lightning/src/util/test_utils.rs +++ b/lightning/src/util/test_utils.rs @@ -17,7 +17,7 @@ use crate::chain::chaininterface; #[cfg(any(test, feature = "_externalize_tests"))] use crate::chain::chaininterface::FEERATE_FLOOR_SATS_PER_KW; use crate::chain::chaininterface::{ConfirmationTarget, TransactionType}; -use crate::chain::chainmonitor::{ChainMonitor, Persist}; +use crate::chain::chainmonitor::{ChainMonitor, MonitorEventSource, Persist}; use crate::chain::channelmonitor::{ ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, MonitorEvent, }; @@ -750,6 +750,10 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { } return self.chain_monitor.release_pending_monitor_events(); } + + fn ack_monitor_event(&self, source: MonitorEventSource) { + self.chain_monitor.ack_monitor_event(source); + } } #[cfg(any(test, feature = "_externalize_tests"))] From ccf17515d6068ff68a7faf9f9f7c4fe782096d30 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Wed, 18 Mar 2026 17:24:04 -0400 Subject: [PATCH 07/20] Add monitor event ids Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. To allow the ChannelManager to ack specific monitor events once they are resolved in upcoming commits, here we give each MonitorEvent a corresponding unique id. It's implemented in such a way that we can delete legacy monitor event code in the future when the new persistent monitor events flag is enabled by default. --- lightning/src/chain/chainmonitor.rs | 2 +- lightning/src/chain/channelmonitor.rs | 200 +++++++++++++----- lightning/src/chain/mod.rs | 8 +- lightning/src/ln/chanmon_update_fail_tests.rs | 6 +- lightning/src/ln/channelmanager.rs | 2 +- lightning/src/util/test_utils.rs | 6 +- 6 files changed, 160 insertions(+), 64 deletions(-) diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index 8832d6fec71..155ffc3caec 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -1639,7 +1639,7 @@ where fn release_pending_monitor_events( &self, - ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)> { + ) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> { for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() { let _ = self.channel_monitor_updated(channel_id, update_id); } diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index aa4ecaaa75c..dabd537beb9 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -184,8 +184,13 @@ impl Readable for ChannelMonitorUpdate { } } -fn push_monitor_event(pending_monitor_events: &mut Vec, event: MonitorEvent) { - pending_monitor_events.push(event); +fn push_monitor_event( + pending_monitor_events: &mut Vec<(u64, MonitorEvent)>, event: MonitorEvent, + next_monitor_event_id: &mut u64, +) { + let id = *next_monitor_event_id; + *next_monitor_event_id += 1; + pending_monitor_events.push((id, event)); } /// An event to be processed by the ChannelManager. @@ -1297,13 +1302,14 @@ pub(crate) struct ChannelMonitorImpl { // Note that because the `event_lock` in `ChainMonitor` is only taken in // block/transaction-connected events and *not* during block/transaction-disconnected events, // we further MUST NOT generate events during block/transaction-disconnection. - pending_monitor_events: Vec, + pending_monitor_events: Vec<(u64, MonitorEvent)>, /// When set, monitor events are retained until explicitly acked rather than cleared on read. /// /// Allows the ChannelManager to reconstruct pending HTLC state by replaying monitor events on /// startup, and make the monitor responsible for both off- and on-chain payment resolution. Will /// be always set once support for this feature is complete. persistent_events_enabled: bool, + next_monitor_event_id: u64, pub(super) pending_events: Vec, pub(super) is_processing_pending_events: bool, @@ -1684,32 +1690,38 @@ pub(crate) fn write_chanmon_internal( writer.write_all(&payment_preimage.0[..])?; } - writer.write_all( - &(channel_monitor - .pending_monitor_events - .iter() - .filter(|ev| match ev { - MonitorEvent::HTLCEvent(_) => true, - MonitorEvent::HolderForceClosed(_) => true, - MonitorEvent::HolderForceClosedWithInfo { .. } => true, - _ => false, - }) - .count() as u64) - .to_be_bytes(), - )?; - for event in channel_monitor.pending_monitor_events.iter() { - match event { - MonitorEvent::HTLCEvent(upd) => { - 0u8.write(writer)?; - upd.write(writer)?; - }, - MonitorEvent::HolderForceClosed(_) => 1u8.write(writer)?, - // `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. To keep - // backwards compatibility, we write a `HolderForceClosed` event along with the - // `HolderForceClosedWithInfo` event. This is deduplicated in the reader. - MonitorEvent::HolderForceClosedWithInfo { .. } => 1u8.write(writer)?, - _ => {}, // Covered in the TLV writes below + if !channel_monitor.persistent_events_enabled { + writer.write_all( + &(channel_monitor + .pending_monitor_events + .iter() + .filter(|(_, ev)| match ev { + MonitorEvent::HTLCEvent(_) => true, + MonitorEvent::HolderForceClosed(_) => true, + MonitorEvent::HolderForceClosedWithInfo { .. } => true, + _ => false, + }) + .count() as u64) + .to_be_bytes(), + )?; + for (_, event) in channel_monitor.pending_monitor_events.iter() { + match event { + MonitorEvent::HTLCEvent(upd) => { + 0u8.write(writer)?; + upd.write(writer)?; + }, + MonitorEvent::HolderForceClosed(_) => 1u8.write(writer)?, + // `HolderForceClosedWithInfo` replaced `HolderForceClosed` in v0.0.122. To keep + // backwards compatibility, we write a `HolderForceClosed` event along with the + // `HolderForceClosedWithInfo` event. This is deduplicated in the reader. + MonitorEvent::HolderForceClosedWithInfo { .. } => 1u8.write(writer)?, + _ => {}, // Covered in the TLV writes below + } } + } else { + // If `persistent_events_enabled` is set, we'll write the events with their event ids in the + // TLV section below. + writer.write_all(&(0u64).to_be_bytes())?; } writer.write_all(&(channel_monitor.pending_events.len() as u64).to_be_bytes())?; @@ -1744,25 +1756,40 @@ pub(crate) fn write_chanmon_internal( // If we have a `HolderForceClosedWithInfo` event, we need to write the `HolderForceClosed` // for backwards compatibility. - let holder_force_closed_compat = channel_monitor.pending_monitor_events.iter().find_map(|ev| { - if let MonitorEvent::HolderForceClosedWithInfo { outpoint, .. } = ev { - Some(MonitorEvent::HolderForceClosed(*outpoint)) - } else { - None - } - }); - let pending_monitor_events_legacy = Iterable( - channel_monitor.pending_monitor_events.iter().chain(holder_force_closed_compat.as_ref()), - ); + let holder_force_closed_compat = + channel_monitor.pending_monitor_events.iter().find_map(|(_, ev)| { + if let MonitorEvent::HolderForceClosedWithInfo { outpoint, .. } = ev { + Some(MonitorEvent::HolderForceClosed(*outpoint)) + } else { + None + } + }); + let pending_monitor_events_legacy = if !channel_monitor.persistent_events_enabled { + Some(Iterable( + channel_monitor + .pending_monitor_events + .iter() + .map(|(_, ev)| ev) + .chain(holder_force_closed_compat.as_ref()), + )) + } else { + None + }; // Only write `persistent_events_enabled` if it's set to true, as it's an even TLV. let persistent_events_enabled = channel_monitor.persistent_events_enabled.then_some(()); + let pending_mon_evs_with_ids = if persistent_events_enabled.is_some() { + Some(Iterable(channel_monitor.pending_monitor_events.iter())) + } else { + None + }; write_tlv_fields!(writer, { (1, channel_monitor.funding_spend_confirmed, option), (2, persistent_events_enabled, option), (3, channel_monitor.htlcs_resolved_on_chain, required_vec), - (5, pending_monitor_events_legacy, required), // Equivalent to required_vec because Iterable also writes as WithoutLength + (4, pending_mon_evs_with_ids, option), + (5, pending_monitor_events_legacy, option), // Equivalent to optional_vec because Iterable also writes as WithoutLength (7, channel_monitor.funding_spend_seen, required), (9, channel_monitor.counterparty_node_id, required), (11, channel_monitor.confirmed_commitment_tx_counterparty_output, option), @@ -1783,6 +1810,7 @@ pub(crate) fn write_chanmon_internal( (37, channel_monitor.funding_seen_onchain, required), (39, channel_monitor.best_block.previous_blocks, required), (41, channel_monitor.funding.contribution, option), + (43, channel_monitor.next_monitor_event_id, required), }); Ok(()) @@ -1969,6 +1997,7 @@ impl ChannelMonitor { payment_preimages: new_hash_map(), pending_monitor_events: Vec::new(), persistent_events_enabled: false, + next_monitor_event_id: 0, pending_events: Vec::new(), is_processing_pending_events: false, @@ -2182,7 +2211,7 @@ impl ChannelMonitor { /// Get the list of HTLCs who's status has been updated on chain. This should be called by /// ChannelManager via [`chain::Watch::release_pending_monitor_events`]. - pub fn get_and_clear_pending_monitor_events(&self) -> Vec { + pub fn get_and_clear_pending_monitor_events(&self) -> Vec<(u64, MonitorEvent)> { self.inner.lock().unwrap().get_and_clear_pending_monitor_events() } @@ -2196,6 +2225,20 @@ impl ChannelMonitor { // TODO: once events have ids, remove the corresponding event here } + /// Copies [`MonitorEvent`] state from `other` into `self`. + /// Used in tests to align transient runtime state before equality comparison after a + /// serialization round-trip. + #[cfg(any(test, feature = "_test_utils"))] + pub fn copy_monitor_event_state(&self, other: &ChannelMonitor) { + let (pending, next_id) = { + let other_inner = other.inner.lock().unwrap(); + (other_inner.pending_monitor_events.clone(), other_inner.next_monitor_event_id) + }; + let mut self_inner = self.inner.lock().unwrap(); + self_inner.pending_monitor_events = pending; + self_inner.next_monitor_event_id = next_id; + } + /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. /// /// For channels featuring anchor outputs, this method will also process [`BumpTransaction`] @@ -3921,7 +3964,7 @@ impl ChannelMonitorImpl { outpoint: funding_outpoint, channel_id: self.channel_id, }; - push_monitor_event(&mut self.pending_monitor_events, event); + push_monitor_event(&mut self.pending_monitor_events, event, &mut self.next_monitor_event_id); } // Although we aren't signing the transaction directly here, the transaction will be signed @@ -4541,12 +4584,16 @@ impl ChannelMonitorImpl { "Failing HTLC from late counterparty commitment update immediately \ (funding spend already confirmed)" ); - self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate { - payment_hash, - payment_preimage: None, - source: source.clone(), - htlc_value_satoshis, - })); + push_monitor_event( + &mut self.pending_monitor_events, + MonitorEvent::HTLCEvent(HTLCUpdate { + payment_hash, + payment_preimage: None, + source: source.clone(), + htlc_value_satoshis, + }), + &mut self.next_monitor_event_id, + ); self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC { commitment_tx_output_idx: None, resolving_txid: Some(confirmed_txid), @@ -4612,10 +4659,14 @@ impl ChannelMonitorImpl { } fn push_monitor_event(&mut self, event: MonitorEvent) { - push_monitor_event(&mut self.pending_monitor_events, event); + push_monitor_event( + &mut self.pending_monitor_events, + event, + &mut self.next_monitor_event_id, + ); } - fn get_and_clear_pending_monitor_events(&mut self) -> Vec { + fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> { let mut ret = Vec::new(); mem::swap(&mut ret, &mut self.pending_monitor_events); ret @@ -5945,7 +5996,7 @@ impl ChannelMonitorImpl { continue; } let duplicate_event = self.pending_monitor_events.iter().any( - |update| if let &MonitorEvent::HTLCEvent(ref upd) = update { + |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == *source } else { false }); if duplicate_event { @@ -5963,7 +6014,7 @@ impl ChannelMonitorImpl { payment_preimage: None, payment_hash: htlc.payment_hash, htlc_value_satoshis: Some(htlc.amount_msat / 1000), - })); + }), &mut self.next_monitor_event_id); } } } @@ -6362,7 +6413,7 @@ impl ChannelMonitorImpl { if let Some((source, payment_hash, amount_msat)) = payment_data { if accepted_preimage_claim { if !self.pending_monitor_events.iter().any( - |update| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) { + |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) { self.onchain_events_awaiting_threshold_conf.push(OnchainEventEntry { txid: tx.compute_txid(), height, @@ -6380,11 +6431,11 @@ impl ChannelMonitorImpl { payment_preimage: Some(payment_preimage), payment_hash, htlc_value_satoshis: Some(amount_msat / 1000), - })); + }), &mut self.next_monitor_event_id); } } else if offered_preimage_claim { if !self.pending_monitor_events.iter().any( - |update| if let &MonitorEvent::HTLCEvent(ref upd) = update { + |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) { self.onchain_events_awaiting_threshold_conf.push(OnchainEventEntry { @@ -6404,7 +6455,7 @@ impl ChannelMonitorImpl { payment_preimage: Some(payment_preimage), payment_hash, htlc_value_satoshis: Some(amount_msat / 1000), - })); + }), &mut self.next_monitor_event_id); } } else { self.onchain_events_awaiting_threshold_conf.retain(|ref entry| { @@ -6770,10 +6821,13 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP let mut best_block_previous_blocks = None; let mut current_funding_contribution = None; let mut persistent_events_enabled: Option<()> = None; + let mut next_monitor_event_id: Option = None; + let mut pending_mon_evs_with_ids: Option> = None; read_tlv_fields!(reader, { (1, funding_spend_confirmed, option), (2, persistent_events_enabled, option), (3, htlcs_resolved_on_chain, optional_vec), + (4, pending_mon_evs_with_ids, optional_vec), (5, pending_monitor_events_legacy, optional_vec), (7, funding_spend_seen, option), (9, counterparty_node_id, option), @@ -6795,6 +6849,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP (37, funding_seen_onchain, (default_value, true)), (39, best_block_previous_blocks, option), // Added and always set in 0.3 (41, current_funding_contribution, option), + (43, next_monitor_event_id, option), }); if let Some(previous_blocks) = best_block_previous_blocks { best_block.previous_blocks = previous_blocks; @@ -6836,6 +6891,22 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP } } + // If persistent events are enabled, use the events with their persisted IDs from TLV 4. + // Otherwise, use the legacy events from TLV 5 and assign sequential IDs. + let (next_monitor_event_id, pending_monitor_events): (u64, Vec<(u64, MonitorEvent)>) = + if persistent_events_enabled.is_some() { + let evs = pending_mon_evs_with_ids.unwrap_or_default() + .into_iter().map(|ev| (ev.0, ev.1)).collect(); + (next_monitor_event_id.unwrap_or(0), evs) + } else if let Some(events) = pending_monitor_events_legacy { + let next_id = next_monitor_event_id.unwrap_or(events.len() as u64); + let evs = events.into_iter().enumerate() + .map(|(i, ev)| (i as u64, ev)).collect(); + (next_id, evs) + } else { + (next_monitor_event_id.unwrap_or(0), Vec::new()) + }; + let channel_parameters = channel_parameters.unwrap_or_else(|| { onchain_tx_handler.channel_parameters().clone() }); @@ -6954,8 +7025,9 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP current_holder_commitment_number, payment_preimages, - pending_monitor_events: pending_monitor_events_legacy.unwrap(), + pending_monitor_events, persistent_events_enabled: persistent_events_enabled.is_some(), + next_monitor_event_id, pending_events, is_processing_pending_events: false, @@ -7004,6 +7076,22 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP } } +/// Deserialization wrapper for reading a `(u64, MonitorEvent)`. +/// Necessary because we can't deserialize a (Readable, MaybeReadable) tuple due to trait +/// conflicts. +struct ReadableIdMonitorEvent(u64, MonitorEvent); + +impl MaybeReadable for ReadableIdMonitorEvent { + fn read(reader: &mut R) -> Result, DecodeError> { + let id: u64 = Readable::read(reader)?; + let event_opt: Option = MaybeReadable::read(reader)?; + match event_opt { + Some(ev) => Ok(Some(ReadableIdMonitorEvent(id, ev))), + None => Ok(None), + } + } +} + #[cfg(test)] pub(super) fn dummy_monitor( channel_id: ChannelId, wrap_signer: impl FnOnce(crate::sign::InMemorySigner) -> S, diff --git a/lightning/src/chain/mod.rs b/lightning/src/chain/mod.rs index ebd1fddaa90..87eb7ccb47c 100644 --- a/lightning/src/chain/mod.rs +++ b/lightning/src/chain/mod.rs @@ -419,6 +419,10 @@ pub trait Watch { /// Returns any monitor events since the last call. Subsequent calls must only return new /// events. /// + /// Each event comes with a corresponding id. Once the event is processed, call + /// [`Watch::ack_monitor_event`] with the corresponding id and channel id. Unacknowledged events + /// will be re-provided by this method after startup. + /// /// Note that after any block- or transaction-connection calls to a [`ChannelMonitor`], no /// further events may be returned here until the [`ChannelMonitor`] has been fully persisted /// to disk. @@ -427,7 +431,7 @@ pub trait Watch { /// [`MonitorEvent::Completed`] here, see [`ChannelMonitorUpdateStatus::InProgress`]. fn release_pending_monitor_events( &self, - ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)>; + ) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)>; /// Acknowledges and removes a [`MonitorEvent`] previously returned by /// [`Watch::release_pending_monitor_events`] by its event ID. @@ -456,7 +460,7 @@ impl + ?Sized, W: Der fn release_pending_monitor_events( &self, - ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)> { + ) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> { self.deref().release_pending_monitor_events() } diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs index 9633800db08..72318db8c93 100644 --- a/lightning/src/ln/chanmon_update_fail_tests.rs +++ b/lightning/src/ln/chanmon_update_fail_tests.rs @@ -5021,7 +5021,7 @@ fn native_async_persist() { let completed_persist = async_chain_monitor.release_pending_monitor_events(); assert_eq!(completed_persist.len(), 1); assert_eq!(completed_persist[0].2.len(), 1); - assert!(matches!(completed_persist[0].2[0], MonitorEvent::Completed { .. })); + assert!(matches!(completed_persist[0].2[0].1, MonitorEvent::Completed { .. })); // Now test two async `ChannelMonitorUpdate`s in flight at once, completing them in-order but // separately. @@ -5069,7 +5069,7 @@ fn native_async_persist() { let completed_persist = async_chain_monitor.release_pending_monitor_events(); assert_eq!(completed_persist.len(), 1); assert_eq!(completed_persist[0].2.len(), 1); - assert!(matches!(completed_persist[0].2[0], MonitorEvent::Completed { .. })); + assert!(matches!(completed_persist[0].2[0].1, MonitorEvent::Completed { .. })); // Finally, test two async `ChanelMonitorUpdate`s in flight at once, completing them // out-of-order and ensuring that no `MonitorEvent::Completed` is generated until they are both @@ -5115,7 +5115,7 @@ fn native_async_persist() { let completed_persist = async_chain_monitor.release_pending_monitor_events(); assert_eq!(completed_persist.len(), 1); assert_eq!(completed_persist[0].2.len(), 1); - if let MonitorEvent::Completed { monitor_update_id, .. } = &completed_persist[0].2[0] { + if let (_, MonitorEvent::Completed { monitor_update_id, .. }) = &completed_persist[0].2[0] { assert_eq!(*monitor_update_id, 4); } else { panic!(); diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index a40c0204acf..6a165525fc8 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -13692,7 +13692,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ for (funding_outpoint, channel_id, mut monitor_events, counterparty_node_id) in pending_monitor_events.drain(..) { - for monitor_event in monitor_events.drain(..) { + for (_event_id, monitor_event) in monitor_events.drain(..) { match monitor_event { MonitorEvent::HTLCEvent(htlc_update) => { let logger = WithContext::from( diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs index f77e9ca39d4..37b63b3ce77 100644 --- a/lightning/src/util/test_utils.rs +++ b/lightning/src/util/test_utils.rs @@ -665,6 +665,7 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { ) .unwrap() .1; + new_monitor.copy_monitor_event_state(&monitor); assert!(new_monitor == monitor); self.latest_monitor_update_id .lock() @@ -726,6 +727,9 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { // it so it doesn't leak into the rest of the test. let failed_back = monitor.inner.lock().unwrap().failed_back_htlc_ids.clone(); new_monitor.inner.lock().unwrap().failed_back_htlc_ids = failed_back; + // The deserialized monitor will reset the monitor event state, so copy it from the live + // monitor before comparing. + new_monitor.copy_monitor_event_state(&monitor); if let Some(chan_id) = self.expect_monitor_round_trip_fail.lock().unwrap().take() { assert_eq!(chan_id, channel_id); assert!(new_monitor != *monitor); @@ -739,7 +743,7 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { fn release_pending_monitor_events( &self, - ) -> Vec<(OutPoint, ChannelId, Vec, PublicKey)> { + ) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> { // Auto-flush pending operations so that the ChannelManager can pick up monitor // completion events. When not in deferred mode the queue is empty so this only // costs a lock acquisition. It ensures standard test helpers (route_payment, etc.) From f744ae24d5d729ae875a34d8b490a2f6deba6128 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Mon, 16 Mar 2026 13:45:21 -0400 Subject: [PATCH 08/20] Ack monitor events immediately Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This will simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. Here for the purposes of merging initial support for persistent monitor events, we ack each immediately after it is received/handled by the ChannelManager, which is equivalent to the behavior we had prior to monitor events becoming persistent. In upcoming work, we'll want to have much more special handling for HTLCUpdate monitor events in particular -- e.g. for outbound payment claim events, we should only ACK the monitor event when the PaymentSent event is processed, until that point we want it to keep being provided back to us on startup. All the other monitor events are trivial to ACK, since they don't need to be re-processed on startup. --- lightning/src/ln/channelmanager.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 6a165525fc8..a5f6fe3c3fb 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -42,6 +42,7 @@ use crate::chain::chaininterface::{ BroadcasterInterface, ConfirmationTarget, FeeEstimator, LowerBoundedFeeEstimator, TransactionType, }; +use crate::chain::chainmonitor::MonitorEventSource; use crate::chain::channelmonitor::{ ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, MonitorEvent, WithChannelMonitor, ANTI_REORG_DELAY, CLTV_CLAIM_BUFFER, HTLC_FAIL_BACK_BUFFER, @@ -13692,7 +13693,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ for (funding_outpoint, channel_id, mut monitor_events, counterparty_node_id) in pending_monitor_events.drain(..) { - for (_event_id, monitor_event) in monitor_events.drain(..) { + for (event_id, monitor_event) in monitor_events.drain(..) { + let monitor_event_source = MonitorEventSource { event_id, channel_id }; match monitor_event { MonitorEvent::HTLCEvent(htlc_update) => { let logger = WithContext::from( @@ -13742,6 +13744,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ completion_update, ); } + self.chain_monitor.ack_monitor_event(monitor_event_source); }, MonitorEvent::HolderForceClosed(_) | MonitorEvent::HolderForceClosedWithInfo { .. } => { @@ -13775,6 +13778,9 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ failed_channels.push((Err(e), counterparty_node_id)); } } + // Channel close monitor events do not need to be replayed on startup because we + // already check the monitors to see if the channel is closed. + self.chain_monitor.ack_monitor_event(monitor_event_source); }, MonitorEvent::CommitmentTxConfirmed(_) => { let per_peer_state = self.per_peer_state.read().unwrap(); @@ -13796,6 +13802,9 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ failed_channels.push((Err(e), counterparty_node_id)); } } + // Channel close monitor events do not need to be replayed on startup because we + // already check the monitors to see if the channel is closed. + self.chain_monitor.ack_monitor_event(monitor_event_source); }, MonitorEvent::Completed { channel_id, monitor_update_id, .. } => { self.channel_monitor_updated( @@ -13803,6 +13812,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ Some(monitor_update_id), &counterparty_node_id, ); + self.chain_monitor.ack_monitor_event(monitor_event_source); }, } } From de8830943b2df301c7656cd9734eb9657dc90a03 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Tue, 17 Mar 2026 15:31:39 -0400 Subject: [PATCH 09/20] Support persistent monitor events Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This will simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. Here we complete work that was built on recent prior commits and actually start re-providing monitor events on startup if they went un-acked during runtime. This isn't actually supported in prod yet, so this new code will run randomly in tests, to ensure we still support the old paths. --- lightning/src/chain/channelmonitor.rs | 61 +++++++++++++++++++++------ lightning/src/ln/channelmanager.rs | 27 +++++++++++- lightning/src/ln/monitor_tests.rs | 3 ++ lightning/src/util/config.rs | 8 ++++ 4 files changed, 86 insertions(+), 13 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index dabd537beb9..bd1dad55f2c 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -1303,6 +1303,11 @@ pub(crate) struct ChannelMonitorImpl { // block/transaction-connected events and *not* during block/transaction-disconnected events, // we further MUST NOT generate events during block/transaction-disconnection. pending_monitor_events: Vec<(u64, MonitorEvent)>, + // `MonitorEvent`s that have been provided to the `ChannelManager` via + // [`ChannelMonitor::get_and_clear_pending_monitor_events`] and are awaiting + // [`ChannelMonitor::ack_monitor_event`] for removal. If an event in this queue is not ack'd, it + // will be re-provided to the `ChannelManager` on startup. + provided_monitor_events: Vec<(u64, MonitorEvent)>, /// When set, monitor events are retained until explicitly acked rather than cleared on read. /// /// Allows the ChannelManager to reconstruct pending HTLC state by replaying monitor events on @@ -1779,7 +1784,12 @@ pub(crate) fn write_chanmon_internal( // Only write `persistent_events_enabled` if it's set to true, as it's an even TLV. let persistent_events_enabled = channel_monitor.persistent_events_enabled.then_some(()); let pending_mon_evs_with_ids = if persistent_events_enabled.is_some() { - Some(Iterable(channel_monitor.pending_monitor_events.iter())) + Some(Iterable( + channel_monitor + .provided_monitor_events + .iter() + .chain(channel_monitor.pending_monitor_events.iter()), + )) } else { None }; @@ -1996,6 +2006,7 @@ impl ChannelMonitor { payment_preimages: new_hash_map(), pending_monitor_events: Vec::new(), + provided_monitor_events: Vec::new(), persistent_events_enabled: false, next_monitor_event_id: 0, pending_events: Vec::new(), @@ -2221,8 +2232,15 @@ impl ChannelMonitor { /// Removes a [`MonitorEvent`] by its event ID, acknowledging that it has been processed. /// Generally called by [`chain::Watch::ack_monitor_event`]. - pub fn ack_monitor_event(&self, _event_id: u64) { - // TODO: once events have ids, remove the corresponding event here + pub fn ack_monitor_event(&self, event_id: u64) { + let inner = &mut *self.inner.lock().unwrap(); + inner.ack_monitor_event(event_id); + } + + /// Enables persistent monitor events mode. When enabled, monitor events are retained until + /// explicitly acked rather than cleared on read. + pub(crate) fn set_persistent_events_enabled(&self, enabled: bool) { + self.inner.lock().unwrap().persistent_events_enabled = enabled; } /// Copies [`MonitorEvent`] state from `other` into `self`. @@ -2230,11 +2248,16 @@ impl ChannelMonitor { /// serialization round-trip. #[cfg(any(test, feature = "_test_utils"))] pub fn copy_monitor_event_state(&self, other: &ChannelMonitor) { - let (pending, next_id) = { + let (provided, pending, next_id) = { let other_inner = other.inner.lock().unwrap(); - (other_inner.pending_monitor_events.clone(), other_inner.next_monitor_event_id) + ( + other_inner.provided_monitor_events.clone(), + other_inner.pending_monitor_events.clone(), + other_inner.next_monitor_event_id, + ) }; let mut self_inner = self.inner.lock().unwrap(); + self_inner.provided_monitor_events = provided; self_inner.pending_monitor_events = pending; self_inner.next_monitor_event_id = next_id; } @@ -4666,10 +4689,23 @@ impl ChannelMonitorImpl { ); } + fn ack_monitor_event(&mut self, event_id: u64) { + self.provided_monitor_events.retain(|(id, _)| *id != event_id); + // If this event was generated prior to a restart, it may be in this queue instead + self.pending_monitor_events.retain(|(id, _)| *id != event_id); + } + fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> { - let mut ret = Vec::new(); - mem::swap(&mut ret, &mut self.pending_monitor_events); - ret + if self.persistent_events_enabled { + let mut ret = Vec::new(); + mem::swap(&mut ret, &mut self.pending_monitor_events); + self.provided_monitor_events.extend(ret.iter().cloned()); + ret + } else { + let mut ret = Vec::new(); + mem::swap(&mut ret, &mut self.pending_monitor_events); + ret + } } /// Gets the set of events that are repeated regularly (e.g. those which RBF bump @@ -5995,8 +6031,8 @@ impl ChannelMonitorImpl { if inbound_htlc_expiry > max_expiry_height { continue; } - let duplicate_event = self.pending_monitor_events.iter().any( - |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { + let duplicate_event = self.pending_monitor_events.iter().chain(self.provided_monitor_events.iter()) + .any(|(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == *source } else { false }); if duplicate_event { @@ -6412,7 +6448,7 @@ impl ChannelMonitorImpl { // HTLC resolution backwards to and figure out whether we learned a preimage from it. if let Some((source, payment_hash, amount_msat)) = payment_data { if accepted_preimage_claim { - if !self.pending_monitor_events.iter().any( + if !self.pending_monitor_events.iter().chain(self.provided_monitor_events.iter()).any( |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) { self.onchain_events_awaiting_threshold_conf.push(OnchainEventEntry { txid: tx.compute_txid(), @@ -6434,7 +6470,7 @@ impl ChannelMonitorImpl { }), &mut self.next_monitor_event_id); } } else if offered_preimage_claim { - if !self.pending_monitor_events.iter().any( + if !self.pending_monitor_events.iter().chain(self.provided_monitor_events.iter()).any( |(_, update)| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) { @@ -7026,6 +7062,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP payment_preimages, pending_monitor_events, + provided_monitor_events: Vec::new(), persistent_events_enabled: persistent_events_enabled.is_some(), next_monitor_event_id, pending_events, diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index a5f6fe3c3fb..c1fd1d31d6d 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -3702,6 +3702,8 @@ impl< our_network_pubkey, current_timestamp, expanded_inbound_key, node_signer.get_receive_auth_key(), secp_ctx.clone(), message_router, logger.clone(), ); + #[cfg(any(test, feature = "_test_utils"))] + let override_persistent_monitor_events = config.override_persistent_monitor_events; ChannelManager { config: RwLock::new(config), @@ -3758,7 +3760,27 @@ impl< logger, - persistent_monitor_events: false, + persistent_monitor_events: { + #[cfg(not(any(test, feature = "_test_utils")))] + { false } + #[cfg(any(test, feature = "_test_utils"))] + { + override_persistent_monitor_events.unwrap_or_else(|| { + use core::hash::{BuildHasher, Hasher}; + match std::env::var("LDK_TEST_PERSISTENT_MON_EVENTS") { + Ok(val) => match val.as_str() { + "1" => true, + "0" => false, + _ => panic!("LDK_TEST_PERSISTENT_MON_EVENTS must be 0 or 1, got: {}", val), + }, + Err(_) => { + let rand_val = std::collections::hash_map::RandomState::new().build_hasher().finish(); + rand_val % 2 == 0 + }, + } + }) + } + }, } } @@ -11778,6 +11800,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ fail_chan!("Already had channel with the new channel_id"); }, hash_map::Entry::Vacant(e) => { + monitor.set_persistent_events_enabled(self.persistent_monitor_events); let monitor_res = self.chain_monitor.watch_channel(monitor.channel_id(), monitor); if let Ok(persist_state) = monitor_res { // There's no problem signing a counterparty's funding transaction if our monitor @@ -11948,6 +11971,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ match chan .funding_signed(&msg, best_block, &self.signer_provider, &self.logger) .and_then(|(funded_chan, monitor)| { + monitor.set_persistent_events_enabled(self.persistent_monitor_events); self.chain_monitor .watch_channel(funded_chan.context.channel_id(), monitor) .map_err(|()| { @@ -12840,6 +12864,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ if let Some(chan) = chan.as_funded_mut() { if let Some(monitor) = monitor_opt { + monitor.set_persistent_events_enabled(self.persistent_monitor_events); let monitor_res = self.chain_monitor.watch_channel(monitor.channel_id(), monitor); if let Ok(persist_state) = monitor_res { diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs index d156f874703..438691b71c7 100644 --- a/lightning/src/ln/monitor_tests.rs +++ b/lightning/src/ln/monitor_tests.rs @@ -3594,6 +3594,9 @@ fn do_test_lost_timeout_monitor_events(confirm_tx: CommitmentType, dust_htlcs: b let mut cfg = test_default_channel_config(); cfg.channel_handshake_config.negotiate_anchors_zero_fee_htlc_tx = true; cfg.channel_handshake_config.negotiate_anchor_zero_fee_commitments = p2a_anchor; + // This test specifically tests lost monitor events, which requires the legacy + // (non-persistent) monitor event behavior. + cfg.override_persistent_monitor_events = Some(false); let cfgs = [Some(cfg.clone()), Some(cfg.clone()), Some(cfg.clone())]; let chanmon_cfgs = create_chanmon_cfgs(3); diff --git a/lightning/src/util/config.rs b/lightning/src/util/config.rs index fa01f8e21b4..d71d37e61f5 100644 --- a/lightning/src/util/config.rs +++ b/lightning/src/util/config.rs @@ -1134,6 +1134,10 @@ pub struct UserConfig { /// /// [`ChannelManager::splice_channel`]: crate::ln::channelmanager::ChannelManager::splice_channel pub reject_inbound_splices: bool, + /// If set to `Some`, overrides the random selection of whether to use persistent monitor + /// events. Only available in tests. + #[cfg(any(test, feature = "_test_utils"))] + pub override_persistent_monitor_events: Option, } impl Default for UserConfig { @@ -1150,6 +1154,8 @@ impl Default for UserConfig { enable_htlc_hold: false, hold_outbound_htlcs_at_next_hop: false, reject_inbound_splices: true, + #[cfg(any(test, feature = "_test_utils"))] + override_persistent_monitor_events: None, } } } @@ -1172,6 +1178,8 @@ impl Readable for UserConfig { hold_outbound_htlcs_at_next_hop: Readable::read(reader)?, enable_htlc_hold: Readable::read(reader)?, reject_inbound_splices: Readable::read(reader)?, + #[cfg(any(test, feature = "_test_utils"))] + override_persistent_monitor_events: None, }) } } From a9d91f548c174d7248c98fab635ea93b5207aea5 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Mon, 30 Mar 2026 11:05:53 -0400 Subject: [PATCH 10/20] Track recent monitor updates in TestChainMonitor And log them in check_added_monitors if it fails. --- lightning/src/ln/functional_test_utils.rs | 15 ++++++++++++++- lightning/src/util/test_utils.rs | 9 +++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs index bbb184d2e48..1107fecd365 100644 --- a/lightning/src/ln/functional_test_utils.rs +++ b/lightning/src/ln/functional_test_utils.rs @@ -1263,7 +1263,20 @@ pub fn check_added_monitors>(node: & if let Some(chain_monitor) = node.chain_monitor() { let mut added_monitors = chain_monitor.added_monitors.lock().unwrap(); let n = added_monitors.len(); - assert_eq!(n, count, "expected {} monitors to be added, not {}", count, n); + if n != count { + let recent = chain_monitor.recent_monitor_updates.lock().unwrap(); + let mut desc = String::new(); + for (i, (chan_id, update)) in recent.iter().take(n).enumerate() { + desc += &format!( + "\n [{}] chan={} update_id={} steps={:?}", + i, chan_id, update.update_id, update.updates + ); + } + panic!( + "expected {} monitors to be added, not {}. Last {} updates (most recent first):{}", + count, n, n, desc + ); + } added_monitors.clear(); } } diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs index 37b63b3ce77..8eb3e6ee7e4 100644 --- a/lightning/src/util/test_utils.rs +++ b/lightning/src/util/test_utils.rs @@ -537,6 +537,8 @@ pub struct TestChainMonitor<'a> { /// deferred operations. This allows tests to control exactly when queued monitor updates /// are applied to the in-memory monitor. pub pause_flush: AtomicBool, + /// Buffer of the last 20 monitor updates, most recent first. + pub recent_monitor_updates: Mutex>, } impl<'a> TestChainMonitor<'a> { pub fn new( @@ -597,6 +599,7 @@ impl<'a> TestChainMonitor<'a> { #[cfg(feature = "std")] write_blocker: Mutex::new(None), pause_flush: AtomicBool::new(false), + recent_monitor_updates: Mutex::new(Vec::new()), } } @@ -695,6 +698,12 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { .or_insert(Vec::new()) .push(update.clone()); + { + let mut recent = self.recent_monitor_updates.lock().unwrap(); + recent.insert(0, (channel_id, update.clone())); + recent.truncate(20); + } + if let Some(exp) = self.expect_channel_force_closed.lock().unwrap().take() { assert_eq!(channel_id, exp.0); assert_eq!(update.updates.len(), 1); From 538e09178fff2c8b3f042772c39e95eedb524780 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Tue, 24 Mar 2026 13:05:25 -0400 Subject: [PATCH 11/20] Persist user channel id in monitors Will be used in upcoming commits when generating MonitorEvents. --- lightning/src/chain/channelmonitor.rs | 10 +++++++++- lightning/src/ln/channel.rs | 2 +- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index bd1dad55f2c..571d3aa6789 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -1226,6 +1226,8 @@ pub(crate) struct ChannelMonitorImpl { /// True if this channel was configured for manual funding broadcasts. Monitors written by /// versions prior to LDK 0.2 load with `false` until a new update persists it. is_manual_broadcast: bool, + /// The user-provided channel ID, used to populate events generated by the monitor. + user_channel_id: Option, /// True once we've observed either funding transaction on-chain. Older monitors prior to LDK 0.2 /// assume this is `true` when absent during upgrade so holder broadcasts aren't gated unexpectedly. /// In manual-broadcast channels we also use this to trigger deferred holder @@ -1821,6 +1823,7 @@ pub(crate) fn write_chanmon_internal( (39, channel_monitor.best_block.previous_blocks, required), (41, channel_monitor.funding.contribution, option), (43, channel_monitor.next_monitor_event_id, required), + (45, channel_monitor.user_channel_id, option), // Added in 0.4 }); Ok(()) @@ -1925,7 +1928,7 @@ impl ChannelMonitor { commitment_transaction_number_obscure_factor: u64, initial_holder_commitment_tx: HolderCommitmentTransaction, best_block: BlockLocator, counterparty_node_id: PublicKey, channel_id: ChannelId, - is_manual_broadcast: bool, + is_manual_broadcast: bool, user_channel_id: u128, ) -> ChannelMonitor { assert!(commitment_transaction_number_obscure_factor <= (1 << 48)); @@ -1976,6 +1979,7 @@ impl ChannelMonitor { pending_funding: vec![], is_manual_broadcast, + user_channel_id: Some(user_channel_id), funding_seen_onchain: false, latest_update_id: 0, @@ -6856,6 +6860,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP let mut funding_seen_onchain = RequiredWrapper(None); let mut best_block_previous_blocks = None; let mut current_funding_contribution = None; + let mut user_channel_id: Option = None; let mut persistent_events_enabled: Option<()> = None; let mut next_monitor_event_id: Option = None; let mut pending_mon_evs_with_ids: Option> = None; @@ -6886,6 +6891,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP (39, best_block_previous_blocks, option), // Added and always set in 0.3 (41, current_funding_contribution, option), (43, next_monitor_event_id, option), + (45, user_channel_id, option), // Added in 0.4 }); if let Some(previous_blocks) = best_block_previous_blocks { best_block.previous_blocks = previous_blocks; @@ -7030,6 +7036,7 @@ impl<'a, 'b, ES: EntropySource, SP: SignerProvider> ReadableArgs<(&'a ES, &'b SP }, pending_funding: pending_funding.unwrap_or(vec![]), is_manual_broadcast: is_manual_broadcast.0.unwrap(), + user_channel_id, // Older monitors prior to LDK 0.2 assume this is `true` when absent // during upgrade so holder broadcasts aren't gated unexpectedly. funding_seen_onchain: funding_seen_onchain.0.unwrap(), @@ -7191,6 +7198,7 @@ pub(super) fn dummy_monitor( dummy_key, channel_id, false, + 0, ) } diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs index edcaacfedc6..1b3ed17c3c9 100644 --- a/lightning/src/ln/channel.rs +++ b/lightning/src/ln/channel.rs @@ -3690,7 +3690,7 @@ trait InitialRemoteCommitmentReceiver { funding.get_holder_selected_contest_delay(), &context.destination_script, &funding.channel_transaction_parameters, funding.is_outbound(), obscure_factor, holder_commitment_tx, best_block, context.counterparty_node_id, context.channel_id(), - context.is_manual_broadcast, + context.is_manual_broadcast, context.get_user_id(), ); channel_monitor.provide_initial_counterparty_commitment_tx( counterparty_initial_commitment_tx.clone(), From f2e42ed9a327129ee8362e0bde83bb836bf52655 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Tue, 24 Mar 2026 14:48:56 -0400 Subject: [PATCH 12/20] Include user channel id in monitor event Processing MonitorEvent::HTLCEvent causes the ChannelManager to call claim_funds_internal, but it will currently pass in None for the user_channel_id parameter. In upcoming commits when we begin generating monitor events for off-chain HTLC claims as well as onchain, we'll want to start using an accurate value instead. --- lightning/src/chain/channelmonitor.rs | 7 +++++++ lightning/src/ln/channelmanager.rs | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 571d3aa6789..a8948497a41 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -262,12 +262,14 @@ pub struct HTLCUpdate { pub(crate) payment_preimage: Option, pub(crate) source: HTLCSource, pub(crate) htlc_value_satoshis: Option, + pub(crate) user_channel_id: Option, } impl_writeable_tlv_based!(HTLCUpdate, { (0, payment_hash, required), (1, htlc_value_satoshis, option), (2, source, required), (4, payment_preimage, option), + (5, user_channel_id, option), }); /// If an output goes from claimable only by us to claimable by us or our counterparty within this @@ -4618,6 +4620,7 @@ impl ChannelMonitorImpl { payment_preimage: None, source: source.clone(), htlc_value_satoshis, + user_channel_id: self.user_channel_id, }), &mut self.next_monitor_event_id, ); @@ -5945,6 +5948,7 @@ impl ChannelMonitorImpl { payment_preimage: None, source, htlc_value_satoshis, + user_channel_id: self.user_channel_id, })); self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC { commitment_tx_output_idx, @@ -6054,6 +6058,7 @@ impl ChannelMonitorImpl { payment_preimage: None, payment_hash: htlc.payment_hash, htlc_value_satoshis: Some(htlc.amount_msat / 1000), + user_channel_id: self.user_channel_id, }), &mut self.next_monitor_event_id); } } @@ -6471,6 +6476,7 @@ impl ChannelMonitorImpl { payment_preimage: Some(payment_preimage), payment_hash, htlc_value_satoshis: Some(amount_msat / 1000), + user_channel_id: self.user_channel_id, }), &mut self.next_monitor_event_id); } } else if offered_preimage_claim { @@ -6495,6 +6501,7 @@ impl ChannelMonitorImpl { payment_preimage: Some(payment_preimage), payment_hash, htlc_value_satoshis: Some(amount_msat / 1000), + user_channel_id: self.user_channel_id, }), &mut self.next_monitor_event_id); } } else { diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index c1fd1d31d6d..bcaf9a2151c 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -13745,7 +13745,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ counterparty_node_id, funding_outpoint, channel_id, - None, + htlc_update.user_channel_id, None, None, ); From 3f63deffa0c00f691493b7558043a30da6032903 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 13:51:17 -0400 Subject: [PATCH 13/20] Pass best block height to outbound_payments::claim_htlc Used in an upcoming commit to insert a pending payment if it's missing on startup and we need to re-claim. --- lightning/src/ln/channelmanager.rs | 3 +++ lightning/src/ln/outbound_payment.rs | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index bcaf9a2151c..fbd112fbba3 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -10193,6 +10193,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ Some(payment_preimage.into()), payment_id, ); + let best_block_height = self.best_block.read().unwrap().height; self.pending_outbound_payments.claim_htlc( payment_id, payment_preimage, @@ -10200,6 +10201,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ session_priv, path, from_onchain, + best_block_height, &mut ev_completion_action, &self.pending_events, &logger, @@ -19990,6 +19992,7 @@ impl< session_priv, path, true, + best_block.height, &mut compl_action, &pending_events, &logger, diff --git a/lightning/src/ln/outbound_payment.rs b/lightning/src/ln/outbound_payment.rs index 7259f60796f..f0a52d50dc5 100644 --- a/lightning/src/ln/outbound_payment.rs +++ b/lightning/src/ln/outbound_payment.rs @@ -2248,7 +2248,7 @@ impl OutboundPayments { #[rustfmt::skip] pub(super) fn claim_htlc( &self, payment_id: PaymentId, payment_preimage: PaymentPreimage, bolt12_invoice: Option, - session_priv: SecretKey, path: Path, from_onchain: bool, ev_completion_action: &mut Option, + session_priv: SecretKey, path: Path, from_onchain: bool, best_block_height: u32, ev_completion_action: &mut Option, pending_events: &Mutex)>>, logger: &WithContext, ) From df3bc78db74df91d6e5354c02d376ebdc3d17918 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 13:53:16 -0400 Subject: [PATCH 14/20] Pass monitor event id to claim_funds_internal Used in an upcoming commit to generate an EventCompletionAction::AckMonitorEvent. --- lightning/src/ln/channelmanager.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index fbd112fbba3..46826a97c83 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -10159,6 +10159,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ next_channel_counterparty_node_id: PublicKey, next_channel_outpoint: OutPoint, next_channel_id: ChannelId, next_user_channel_id: Option, attribution_data: Option, send_timestamp: Option, + monitor_event_id: Option, ) { let startup_replay = !self.background_events_processed_since_startup.load(Ordering::Acquire); @@ -12782,6 +12783,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ Some(next_user_channel_id), msg.attribution_data, send_timestamp, + None, ); Ok(()) @@ -13750,6 +13752,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ htlc_update.user_channel_id, None, None, + Some(event_id), ); } else { log_trace!(logger, "Failing HTLC from our monitor"); @@ -20813,6 +20816,7 @@ impl< downstream_user_channel_id, None, None, + None, ); } From cb071ef6190d0c3b3eb70a99c56fc5d11a3934bc Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 14:50:30 -0400 Subject: [PATCH 15/20] Stop hardcoding from_onchain in monitor ev claim_funds In upcoming commits, we'll be generating monitor events for off-chain claims as well as on-chain. As a small prefactor, calculate the from_onchain value rather than hardcoding it to true. --- lightning/src/ln/channelmanager.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 46826a97c83..142454470d6 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -13738,6 +13738,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ "Claiming HTLC with preimage {} from our monitor", preimage ); + let from_onchain = self + .per_peer_state + .read() + .unwrap() + .get(&counterparty_node_id) + .map_or(true, |peer_state_mtx| { + !peer_state_mtx + .lock() + .unwrap() + .channel_by_id + .contains_key(&channel_id) + }); // Claim the funds from the previous hop, if there is one. Because this is in response to a // chain event, no attribution data is available. self.claim_funds_internal( @@ -13745,7 +13757,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ preimage, htlc_update.htlc_value_satoshis.map(|v| v * 1000), None, - true, + from_onchain, counterparty_node_id, funding_outpoint, channel_id, From 5c6f09d65ee4e1b1b7c2fde4c0cc1e604463943e Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 12:22:38 -0400 Subject: [PATCH 16/20] Add EventCompletionAction::AckMonitorEvent If ChannelManager::persistent_monitor_events is enabled, we may want to avoid acking a monitor event until after an Event is processed by the user. In upcoming commits, we'll use this to ensure a MonitorEvent::HTLCEvent will keep being re-provided back to us until after an Event::PaymentSent is processed. --- lightning/src/ln/channelmanager.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 142454470d6..98eb629e9cb 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -1635,6 +1635,11 @@ pub(crate) enum EventCompletionAction { /// fully-resolved in the [`ChannelMonitor`], which we do via this action. /// Note that this action will be dropped on downgrade to LDK prior to 0.2! ReleasePaymentCompleteChannelMonitorUpdate(PaymentCompleteUpdate), + /// If [`ChannelManager::persistent_monitor_events`] is enabled, we may want to avoid acking a + /// monitor event via [`Watch::ack_monitor_event`] until after an [`Event`] is processed by the + /// user. For example, we may want a [`MonitorEvent::HTLCEvent`] to keep being re-provided to us + /// until after an [`Event::PaymentSent`] is processed. + AckMonitorEvent { event_id: MonitorEventSource }, } impl_writeable_tlv_based_enum!(EventCompletionAction, (0, ReleaseRAAChannelMonitorUpdate) => { @@ -1646,6 +1651,9 @@ impl_writeable_tlv_based_enum!(EventCompletionAction, } ChannelId::v1_from_funding_outpoint(channel_funding_outpoint.unwrap()) })), + }, + (3, AckMonitorEvent) => { + (1, event_id, required), } {1, ReleasePaymentCompleteChannelMonitorUpdate} => (), ); @@ -15659,6 +15667,9 @@ impl< } } }, + EventCompletionAction::AckMonitorEvent { event_id } => { + self.chain_monitor.ack_monitor_event(event_id); + }, } } } From 7363f7ce309c972da796dd0a0178126259ce0d99 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 13:56:33 -0400 Subject: [PATCH 17/20] Persistent mon events for off-chain outbound claims Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This will simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. In recent work, we added support for keeping monitor events around until they are explicitly acked by the ChannelManager, but would always ack monitor events immediately, which preserved the previous behavior and didn't break any tests. Up until this point, we only generated HTLC monitor events when a payment was claimed/failed on-chain. In this commit, we start generating persistent monitor events whenever a payment is claimed *off*-chain, specifically when new latest holder commitment data is provided to the monitor. For the purpose of making incremental progress on this feature, these events will be a no-op and/or continue to be acked immediately except in the narrow case of an off-chain outbound payment claim. HTLC forward claim monitor events will be a no-op, and on-chain outbound payment claim events continue to be acked immediately. Off-chain outbound payment claims, however, now have monitor events generated for them that will not be acked by the ChannelManager until the PaymentSent event is processed by the user. This also allows us to stop blocking the RAA monitor update that removes the preimage, because the purpose of that behavior was to ensure the user got a PaymentSent event and the monitor event now serves that purpose instead. --- lightning/src/chain/channelmonitor.rs | 52 ++++++++--- lightning/src/ln/blinded_payment_tests.rs | 4 +- lightning/src/ln/chanmon_update_fail_tests.rs | 25 +++-- lightning/src/ln/channelmanager.rs | 91 ++++++++++++++----- lightning/src/ln/functional_test_utils.rs | 12 ++- lightning/src/ln/functional_tests.rs | 20 +++- lightning/src/ln/invoice_utils.rs | 2 +- lightning/src/ln/monitor_tests.rs | 7 +- lightning/src/ln/outbound_payment.rs | 10 ++ lightning/src/ln/payment_tests.rs | 30 ++++-- lightning/src/ln/quiescence_tests.rs | 33 ++++++- lightning/src/ln/reload_tests.rs | 22 +++-- lightning/src/ln/shutdown_tests.rs | 5 +- 13 files changed, 240 insertions(+), 73 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index a8948497a41..5d7968c5db5 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -2249,6 +2249,10 @@ impl ChannelMonitor { self.inner.lock().unwrap().persistent_events_enabled = enabled; } + pub(crate) fn has_pending_event_for_htlc(&self, source: &HTLCSource) -> bool { + self.inner.lock().unwrap().has_pending_event_for_htlc(source) + } + /// Copies [`MonitorEvent`] state from `other` into `self`. /// Used in tests to align transient runtime state before equality comparison after a /// serialization round-trip. @@ -3842,20 +3846,34 @@ impl ChannelMonitorImpl { self.prev_holder_htlc_data = Some(htlc_data); for (claimed_htlc_id, claimed_preimage) in claimed_htlcs { - #[cfg(debug_assertions)] - { - let cur_counterparty_htlcs = self - .funding - .counterparty_claimable_outpoints - .get(&self.funding.current_counterparty_commitment_txid.unwrap()) - .unwrap(); - assert!(cur_counterparty_htlcs.iter().any(|(_, source_opt)| { + let htlc_opt = self + .funding + .counterparty_claimable_outpoints + .get(&self.funding.current_counterparty_commitment_txid.unwrap()) + .unwrap() + .iter() + .find_map(|(htlc, source_opt)| { if let Some(source) = source_opt { - SentHTLCId::from_source(source) == *claimed_htlc_id - } else { - false + if SentHTLCId::from_source(source) == *claimed_htlc_id { + return Some((htlc, source)); + } } - })); + None + }); + debug_assert!(htlc_opt.is_some()); + if self.persistent_events_enabled { + if let Some((htlc, source)) = htlc_opt { + // If persistent_events_enabled is set, the ChannelMonitor is responsible for providing + // off-chain resolutions of HTLCs to the ChannelManager, will re-provide this event on + // startup until it is explicitly acked. + self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate { + payment_hash: htlc.payment_hash, + payment_preimage: Some(*claimed_preimage), + source: *source.clone(), + htlc_value_satoshis: Some(htlc.amount_msat), + user_channel_id: self.user_channel_id, + })); + } } self.counterparty_fulfilled_htlcs.insert(*claimed_htlc_id, *claimed_preimage); } @@ -4702,6 +4720,16 @@ impl ChannelMonitorImpl { self.pending_monitor_events.retain(|(id, _)| *id != event_id); } + fn has_pending_event_for_htlc(&self, htlc: &HTLCSource) -> bool { + let htlc_id = SentHTLCId::from_source(htlc); + self.pending_monitor_events.iter().any(|(_, ev)| { + if let MonitorEvent::HTLCEvent(upd) = ev { + return htlc_id == SentHTLCId::from_source(&upd.source); + } + false + }) + } + fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> { if self.persistent_events_enabled { let mut ret = Vec::new(); diff --git a/lightning/src/ln/blinded_payment_tests.rs b/lightning/src/ln/blinded_payment_tests.rs index 32c0709ed5c..2308f357937 100644 --- a/lightning/src/ln/blinded_payment_tests.rs +++ b/lightning/src/ln/blinded_payment_tests.rs @@ -856,7 +856,7 @@ fn do_blinded_intercept_payment(intercept_node_fails: bool) { do_claim_payment_along_route( ClaimAlongRouteArgs::new(&nodes[0], &[&[&nodes[1], &nodes[2]]], payment_preimage) ); - expect_payment_sent(&nodes[0], payment_preimage, Some(Some(1000)), true, true); + expect_payment_sent!(nodes[0], payment_preimage, Some(1000)); } #[test] @@ -1401,7 +1401,7 @@ fn conditionally_round_fwd_amt() { let mut args = ClaimAlongRouteArgs::new(&nodes[0], &expected_route[..], payment_preimage) .allow_1_msat_fee_overpay(); let expected_fee = pass_claimed_payment_along_route(args); - expect_payment_sent(&nodes[0], payment_preimage, Some(Some(expected_fee)), true, true); + expect_payment_sent!(nodes[0], payment_preimage, Some(expected_fee)); } #[test] diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs index 72318db8c93..4ef20358848 100644 --- a/lightning/src/ln/chanmon_update_fail_tests.rs +++ b/lightning/src/ln/chanmon_update_fail_tests.rs @@ -2108,7 +2108,7 @@ fn monitor_update_claim_fail_no_response() { let mut bs_updates = get_htlc_update_msgs(&nodes[1], &node_a_id); nodes[0].node.handle_update_fulfill_htlc(node_b_id, bs_updates.update_fulfill_htlcs.remove(0)); do_commitment_signed_dance(&nodes[0], &nodes[1], &bs_updates.commitment_signed, false, false); - expect_payment_sent!(nodes[0], payment_preimage_1); + expect_payment_sent!(&nodes[0], payment_preimage_1); claim_payment(&nodes[0], &[&nodes[1]], payment_preimage_2); } @@ -3450,7 +3450,10 @@ fn do_test_blocked_chan_preimage_release(completion_mode: BlockedUpdateComplMode let node_cfgs = create_node_cfgs(3, &chanmon_cfgs); let persister; let new_chain_mon; - let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]); + let mut cfg = test_default_channel_config(); + // If persistent_monitor_events is enabed, monitor updates will never be blocked. + cfg.override_persistent_monitor_events = Some(false); + let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, Some(cfg.clone()), Some(cfg)]); let nodes_1_reload; let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs); @@ -3763,7 +3766,7 @@ fn do_test_inverted_mon_completion_order( ); // Finally, check that the payment was, ultimately, seen as sent by node A. - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } #[test] @@ -4256,7 +4259,7 @@ fn do_test_glacial_peer_cant_hang(hold_chan_a: bool) { nodes[0].node.handle_update_fulfill_htlc(node_b_id, update_fulfill); let commitment = &a_update[0].commitment_signed; do_commitment_signed_dance(&nodes[0], &nodes[1], commitment, false, false); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(nodes[0], payment_preimage); expect_payment_forwarded!(nodes[1], nodes[0], nodes[2], Some(1000), false, false); pass_along_path( @@ -4936,6 +4939,7 @@ fn native_async_persist() { let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); let nodes = create_network(2, &node_cfgs, &node_chanmgrs); + let persistent_monitor_events = nodes[0].node.test_persistent_monitor_events_enabled(); let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1); @@ -5081,7 +5085,16 @@ fn native_async_persist() { assert_eq!(update_status, ChannelMonitorUpdateStatus::InProgress); persist_futures.poll_futures(); - assert_eq!(async_chain_monitor.release_pending_monitor_events().len(), 0); + let events = async_chain_monitor.release_pending_monitor_events(); + if persistent_monitor_events { + // With persistent monitor events, the LatestHolderCommitmentTXInfo update containing + // claimed_htlcs generates an HTLCEvent with the preimage. + assert_eq!(events.len(), 1); + assert_eq!(events[0].2.len(), 1); + assert!(matches!(events[0].2[0].1, MonitorEvent::HTLCEvent(..))); + } else { + assert!(events.is_empty()); + } let pending_writes = kv_store.list_pending_async_writes( CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE, @@ -5223,7 +5236,7 @@ fn test_mpp_claim_to_holding_cell() { let claims = vec![(b_claim_msgs, node_b_id), (c_claim_msgs, node_c_id)]; pass_claimed_payment_along_route_from_ev(250_000, claims, args); - expect_payment_sent(&nodes[0], preimage_1, None, true, true); + expect_payment_sent!(nodes[0], preimage_1); expect_and_process_pending_htlcs(&nodes[3], false); expect_payment_claimable!(nodes[3], paymnt_hash_2, payment_secret_2, 400_000); diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 98eb629e9cb..f18e207893c 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -10189,11 +10189,17 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ }; Some(EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(release)) } else { - Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate { - channel_funding_outpoint: Some(next_channel_outpoint), - channel_id: next_channel_id, - counterparty_node_id: path.hops[0].pubkey, - }) + if self.persistent_monitor_events { + monitor_event_id.map(|event_id| EventCompletionAction::AckMonitorEvent { + event_id: MonitorEventSource { channel_id: next_channel_id, event_id }, + }) + } else { + Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate { + channel_funding_outpoint: Some(next_channel_outpoint), + channel_id: next_channel_id, + counterparty_node_id: path.hops[0].pubkey, + }) + } }; let logger = WithContext::for_payment( &self.logger, @@ -10215,6 +10221,25 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ &self.pending_events, &logger, ); + + if matches!( + ev_completion_action, + Some(EventCompletionAction::AckMonitorEvent { .. }) + ) { + // If the `PaymentSent` for this redundant claim is still pending, add the event + // completion action here to ensure the `PaymentSent` will always be regenerated until it + // is processed by the user -- as long as the monitor event corresponding to this + // completion action is not acked, it will continue to be re-provided on startup. + let mut pending_events = self.pending_events.lock().unwrap(); + for (ev, act_opt) in pending_events.iter_mut() { + let found_payment_sent = matches!(ev, Event::PaymentSent { payment_id: Some(id), .. } if *id == payment_id); + if found_payment_sent && act_opt.is_none() { + *act_opt = ev_completion_action.take(); + break; + } + } + } + // If an event was generated, `claim_htlc` set `ev_completion_action` to None, if // not, we should go ahead and run it now (as the claim was duplicative), at least // if a PaymentClaimed event with the same action isn't already pending. @@ -13024,6 +13049,13 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ }) } + /// Returns `true` if `ChannelManager::persistent_monitor_events` is enabled. This flag will only + /// be set randomly in tests for now. + #[cfg(any(test, feature = "_test_utils"))] + pub fn test_persistent_monitor_events_enabled(&self) -> bool { + self.persistent_monitor_events + } + #[cfg(any(test, feature = "_test_utils"))] pub(crate) fn test_raa_monitor_updates_held( &self, counterparty_node_id: PublicKey, channel_id: ChannelId, @@ -13758,22 +13790,29 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ .channel_by_id .contains_key(&channel_id) }); - // Claim the funds from the previous hop, if there is one. Because this is in response to a - // chain event, no attribution data is available. - self.claim_funds_internal( - htlc_update.source, - preimage, - htlc_update.htlc_value_satoshis.map(|v| v * 1000), - None, - from_onchain, - counterparty_node_id, - funding_outpoint, - channel_id, - htlc_update.user_channel_id, - None, - None, - Some(event_id), - ); + let we_are_sender = + matches!(htlc_update.source, HTLCSource::OutboundRoute { .. }); + if from_onchain | we_are_sender { + // Claim the funds from the previous hop, if there is one. In the future we can + // store attribution data in the `ChannelMonitor` and provide it here. + self.claim_funds_internal( + htlc_update.source, + preimage, + htlc_update.htlc_value_satoshis.map(|v| v * 1000), + None, + from_onchain, + counterparty_node_id, + funding_outpoint, + channel_id, + htlc_update.user_channel_id, + None, + None, + Some(event_id), + ); + } + if from_onchain | !we_are_sender { + self.chain_monitor.ack_monitor_event(monitor_event_source); + } } else { log_trace!(logger, "Failing HTLC from our monitor"); let failure_reason = LocalHTLCFailureReason::OnChainTimeout; @@ -13793,8 +13832,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ failure_type, completion_update, ); + self.chain_monitor.ack_monitor_event(monitor_event_source); } - self.chain_monitor.ack_monitor_event(monitor_event_source); }, MonitorEvent::HolderForceClosed(_) | MonitorEvent::HolderForceClosedWithInfo { .. } => { @@ -19365,6 +19404,14 @@ impl< break; } } + if persistent_monitor_events { + // This will not be necessary once we have persistent events for HTLC failures, we + // can delete this whole loop and wait to re-process the pending monitor events + // rather than failing them proactively below. + if monitor.has_pending_event_for_htlc(&channel_htlc_source) { + found_htlc = true; + } + } if !found_htlc { // If we have some HTLCs in the channel which are not present in the newer // ChannelMonitor, they have been removed and should be failed back to diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs index 1107fecd365..01b32a4bf9c 100644 --- a/lightning/src/ln/functional_test_utils.rs +++ b/lightning/src/ln/functional_test_utils.rs @@ -3078,7 +3078,7 @@ macro_rules! expect_payment_sent { $expected_payment_preimage, $expected_fee_msat_opt.map(|o| Some(o)), $expect_paths, - true, + if $node.node.test_persistent_monitor_events_enabled() { false } else { true }, ) }; } @@ -4241,7 +4241,15 @@ pub fn claim_payment_along_route( do_claim_payment_along_route(args) + expected_extra_total_fees_msat; if !skip_last { - expect_payment_sent!(origin_node, payment_preimage, Some(expected_total_fee_msat)) + let expect_post_ev_mon_update = + if origin_node.node.test_persistent_monitor_events_enabled() { false } else { true }; + expect_payment_sent( + origin_node, + payment_preimage, + Some(Some(expected_total_fee_msat)), + true, + expect_post_ev_mon_update, + ) } else { (None, Vec::new()) } diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs index 7393f354010..12b6beb0621 100644 --- a/lightning/src/ln/functional_tests.rs +++ b/lightning/src/ln/functional_tests.rs @@ -1361,6 +1361,10 @@ pub fn do_test_multiple_package_conflicts(p2a_anchor: bool) { }; assert_eq!(updates.update_fulfill_htlcs.len(), 1); nodes[0].node.handle_update_fulfill_htlc(node_b_id, updates.update_fulfill_htlcs.remove(0)); + if nodes[0].node.test_persistent_monitor_events_enabled() { + // If persistent_monitor_events is enabled, the RAA monitor update is not blocked. + check_added_monitors(&nodes[0], 1); + } do_commitment_signed_dance(&nodes[0], &nodes[1], &updates.commitment_signed, false, false); expect_payment_sent!(nodes[0], preimage_2); @@ -2675,7 +2679,10 @@ pub fn test_simple_peer_disconnect() { _ => panic!("Unexpected event"), } } - check_added_monitors(&nodes[0], 1); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); claim_payment(&nodes[0], &[&nodes[1], &nodes[2]], payment_preimage_4); fail_payment(&nodes[0], &[&nodes[1], &nodes[2]], payment_hash_6); @@ -4300,7 +4307,7 @@ pub fn test_duplicate_payment_hash_one_failure_one_success() { nodes[0].node.handle_update_fulfill_htlc(node_b_id, updates.update_fulfill_htlcs.remove(0)); do_commitment_signed_dance(&nodes[0], &nodes[1], &updates.commitment_signed, false, false); - expect_payment_sent(&nodes[0], our_payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], our_payment_preimage); } #[xtest(feature = "_externalize_tests")] @@ -8579,7 +8586,9 @@ pub fn test_inconsistent_mpp_params() { pass_along_path(&nodes[0], path_b, real_amt, hash, Some(payment_secret), event, true, None); do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], &[path_a, path_b], preimage)); - expect_payment_sent(&nodes[0], preimage, Some(None), true, true); + let expect_post_ev_mon_update = + if nodes[0].node.test_persistent_monitor_events_enabled() { false } else { true }; + expect_payment_sent(&nodes[0], preimage, Some(None), true, expect_post_ev_mon_update); } #[xtest(feature = "_externalize_tests")] @@ -9941,7 +9950,10 @@ fn do_test_multi_post_event_actions(do_reload: bool) { let chanmon_cfgs = create_chanmon_cfgs(3); let node_cfgs = create_node_cfgs(3, &chanmon_cfgs); let (persister, chain_monitor); - let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]); + let mut cfg = test_default_channel_config(); + // If persistent_monitor_events is enabled, RAAs will not be blocked on events. + cfg.override_persistent_monitor_events = Some(false); + let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[Some(cfg), None, None]); let node_a_reload; let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs); diff --git a/lightning/src/ln/invoice_utils.rs b/lightning/src/ln/invoice_utils.rs index 564203bf524..daaa8d9fde5 100644 --- a/lightning/src/ln/invoice_utils.rs +++ b/lightning/src/ln/invoice_utils.rs @@ -1347,7 +1347,7 @@ mod test { &[&[&nodes[fwd_idx]]], payment_preimage, )); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } #[test] diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs index 438691b71c7..043ecddf7b5 100644 --- a/lightning/src/ln/monitor_tests.rs +++ b/lightning/src/ln/monitor_tests.rs @@ -3250,14 +3250,15 @@ fn test_event_replay_causing_monitor_replay() { // Now process the `PaymentSent` to get the final RAA `ChannelMonitorUpdate`, checking that it // resulted in a `ChannelManager` persistence request. nodes[0].node.get_and_clear_needs_persistence(); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true /* expected post-event monitor update*/); + expect_payment_sent!(nodes[0], payment_preimage); assert!(nodes[0].node.get_and_clear_needs_persistence()); let serialized_monitor = get_monitor!(nodes[0], chan.2).encode(); reload_node!(nodes[0], &serialized_channel_manager, &[&serialized_monitor], persister, new_chain_monitor, node_deserialized); // Expect the `PaymentSent` to get replayed, this time without the duplicate monitor update - expect_payment_sent(&nodes[0], payment_preimage, None, false, false /* expected post-event monitor update*/); + let per_path_evs = if nodes[0].node.test_persistent_monitor_events_enabled() { true } else { false }; + expect_payment_sent(&nodes[0], payment_preimage, None, per_path_evs, false /* expected post-event monitor update*/); } #[test] @@ -3555,7 +3556,7 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool, p2a_anchor: bo let mut reconnect_args = ReconnectArgs::new(&nodes[0], &nodes[1]); reconnect_args.pending_cell_htlc_claims = (1, 0); reconnect_nodes(reconnect_args); - expect_payment_sent(&nodes[0], preimage_a, None, true, true); + expect_payment_sent!(&nodes[0], preimage_a); } #[test] diff --git a/lightning/src/ln/outbound_payment.rs b/lightning/src/ln/outbound_payment.rs index f0a52d50dc5..f41a4f06459 100644 --- a/lightning/src/ln/outbound_payment.rs +++ b/lightning/src/ln/outbound_payment.rs @@ -2256,6 +2256,16 @@ impl OutboundPayments { { let mut session_priv_bytes = [0; 32]; session_priv_bytes.copy_from_slice(&session_priv[..]); + + if from_onchain { + // If we are re-processing this claim from a `MonitorEvent` and the `ChannelManager` is + // outdated and has no idea about the payment, we may need to re-insert here to ensure a + // `PaymentSent` event gets generated. + self.insert_from_monitor_on_startup( + payment_id, payment_preimage.into(), session_priv_bytes, &path, best_block_height, logger + ); + } + let mut outbounds = self.pending_outbound_payments.lock().unwrap(); let mut pending_events = pending_events.lock().unwrap(); if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) { diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs index 445fa1a1eea..59725f62bc1 100644 --- a/lightning/src/ln/payment_tests.rs +++ b/lightning/src/ln/payment_tests.rs @@ -2462,7 +2462,11 @@ fn do_test_intercepted_payment(test: InterceptTest) { }, _ => panic!("Unexpected event"), } - check_added_monitors(&nodes[0], 1); + if nodes[0].node.test_persistent_monitor_events_enabled() { + check_added_monitors(&nodes[0], 0); + } else { + check_added_monitors(&nodes[0], 1); + } } else if test == InterceptTest::Timeout { let mut block = create_dummy_block(nodes[0].best_block_hash(), 42, Vec::new()); connect_block(&nodes[0], &block); @@ -2667,7 +2671,7 @@ fn do_accept_underpaying_htlcs_config(num_mpp_parts: usize) { let total_fee_msat = pass_claimed_payment_along_route(args); // The sender doesn't know that the penultimate hop took an extra fee. let amt = total_fee_msat - skimmed_fee_msat * num_mpp_parts as u64; - expect_payment_sent(&nodes[0], payment_preimage, Some(Some(amt)), true, true); + expect_payment_sent!(nodes[0], payment_preimage, Some(amt)); } #[derive(PartialEq)] @@ -4848,7 +4852,7 @@ fn do_test_custom_tlvs_consistency( ClaimAlongRouteArgs::new(&nodes[0], &[path_a, &[&nodes[2], &nodes[3]]], preimage) .with_custom_tlvs(expected_tlvs), ); - expect_payment_sent(&nodes[0], preimage, Some(Some(2000)), true, true); + expect_payment_sent!(nodes[0], preimage, Some(2000)); } else { // Expect fail back let expected_destinations = [HTLCHandlingFailureType::Receive { payment_hash: hash }]; @@ -5656,7 +5660,10 @@ fn do_bolt11_multi_node_mpp(use_bolt11_pay: bool) { } let payment_sent = nodes[0].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[0], 1); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); assert_eq!(payment_sent.len(), 2, "{payment_sent:?}"); if let Event::PaymentSent { payment_id, payment_hash, amount_msat, fee_paid_msat, .. } = @@ -5685,7 +5692,10 @@ fn do_bolt11_multi_node_mpp(use_bolt11_pay: bool) { } let payment_sent = nodes[1].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[1], 1); + check_added_monitors( + &nodes[1], + if nodes[1].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); assert_eq!(payment_sent.len(), 2, "{payment_sent:?}"); if let Event::PaymentSent { payment_id, payment_hash, amount_msat, fee_paid_msat, .. } = @@ -5941,7 +5951,10 @@ fn bolt11_multi_node_mpp_with_retry() { } let payment_sent_a = nodes[0].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[0], 1); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); assert_eq!(payment_sent_a.len(), 2, "{payment_sent_a:?}"); if let Event::PaymentSent { payment_id, payment_hash, amount_msat, .. } = &payment_sent_a[0] { @@ -5966,7 +5979,10 @@ fn bolt11_multi_node_mpp_with_retry() { } let payment_sent_b = nodes[1].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[1], 1); + check_added_monitors( + &nodes[1], + if nodes[1].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); assert_eq!(payment_sent_b.len(), 2, "{payment_sent_b:?}"); if let Event::PaymentSent { payment_id, payment_hash, amount_msat, .. } = &payment_sent_b[0] { diff --git a/lightning/src/ln/quiescence_tests.rs b/lightning/src/ln/quiescence_tests.rs index 495a1622522..aa587e4fa8c 100644 --- a/lightning/src/ln/quiescence_tests.rs +++ b/lightning/src/ln/quiescence_tests.rs @@ -259,7 +259,7 @@ fn test_quiescence_waits_for_async_signer_and_monitor_update() { assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); // We have two updates pending: - { + if !nodes[0].node.test_persistent_monitor_events_enabled() { let test_chain_mon = &nodes[0].chain_monitor; let (_, latest_update) = test_chain_mon.latest_monitor_update_id.lock().unwrap().get(&chan_id).unwrap().clone(); @@ -280,6 +280,27 @@ fn test_quiescence_waits_for_async_signer_and_monitor_update() { } else { panic!("{events:?}"); } + } else { + // In the persistent monitor events path, we don't block the RAA monitor update, so + // `expect_post_ev_mon_update` is false here + expect_payment_sent(&nodes[0], preimage, None, false, false); + // The latest commitment transaction update from the the `revoke_and_ack` was previously + // blocked via `InProgress`, so update that here, which will unblock the commitment secret + // update from the RAA + let test_chain_mon = &nodes[0].chain_monitor; + let (_, latest_update) = + test_chain_mon.latest_monitor_update_id.lock().unwrap().get(&chan_id).unwrap().clone(); + let chain_monitor = &nodes[0].chain_monitor.chain_monitor; + // Complete the held update, which releases the commitment secret update, which releases the + // `PaymentPathSuccessful` event + chain_monitor.channel_monitor_updated(chan_id, latest_update).unwrap(); + check_added_monitors(&nodes[0], 1); + let events = nodes[0].node.get_and_clear_pending_events(); + assert_eq!(events.len(), 1); + if let Event::PaymentPathSuccessful { .. } = &events[0] { + } else { + panic!("{events:?}"); + } } // With the updates completed, we can now become quiescent. @@ -438,7 +459,10 @@ fn quiescence_updates_go_to_holding_cell(fail_htlc: bool) { } else { assert!(events.iter().find(|e| matches!(e, Event::PaymentSent { .. })).is_some()); assert!(events.iter().find(|e| matches!(e, Event::PaymentPathSuccessful { .. })).is_some()); - check_added_monitors(&nodes[0], 1); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }, + ); } nodes[0].node.process_pending_htlc_forwards(); expect_payment_claimable!(nodes[0], payment_hash1, payment_secret1, payment_amount); @@ -467,7 +491,7 @@ fn quiescence_updates_go_to_holding_cell(fail_htlc: bool) { let conditions = PaymentFailedConditions::new(); expect_payment_failed_conditions(&nodes[1], payment_hash1, true, conditions); } else { - expect_payment_sent(&nodes[1], payment_preimage1, None, true, true); + expect_payment_sent!(&nodes[1], payment_preimage1); } } @@ -673,6 +697,9 @@ fn do_test_quiescence_during_disconnection(with_pending_claim: bool, propose_dis assert_eq!(bs_raa_stfu.len(), 2); if let MessageSendEvent::SendRevokeAndACK { msg, .. } = &bs_raa_stfu[0] { nodes[0].node.handle_revoke_and_ack(node_b_id, &msg); + if nodes[0].node.test_persistent_monitor_events_enabled() { + check_added_monitors(&nodes[0], 1); + } expect_payment_sent!(&nodes[0], preimage); } else { panic!("Unexpected first message {bs_raa_stfu:?}"); diff --git a/lightning/src/ln/reload_tests.rs b/lightning/src/ln/reload_tests.rs index 9da90d95109..7fbc62948a9 100644 --- a/lightning/src/ln/reload_tests.rs +++ b/lightning/src/ln/reload_tests.rs @@ -1229,9 +1229,11 @@ fn test_mpp_claim_htlc_fulfills_unblocked_on_reload() { }, _ => panic!("Unexpected event: {:?}", claim_events[0]), } - // The `PaymentSent` event above releases the monitor update that nodes[0] held after the final - // channel A startup revocation. - check_added_monitors(&nodes[0], 1); + if !nodes[0].node.test_persistent_monitor_events_enabled() { + // The `PaymentSent` event above releases the monitor update that nodes[0] held after the final + // channel A startup revocation. + check_added_monitors(&nodes[0], 1); + } // Handling `PaymentClaimed` releases channel B's held revocation update and then the fulfill // that was waiting behind it. check_added_monitors(&nodes[1], 2); @@ -1648,7 +1650,7 @@ fn do_manager_persisted_pre_outbound_edge_forward(intercept_htlc: bool) { expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat, None, nodes[2].node.get_our_node_id()); let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]]; do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], path, payment_preimage)); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } #[test] @@ -1706,7 +1708,7 @@ fn test_manager_persisted_post_outbound_edge_forward() { expect_payment_claimable!(nodes[2], payment_hash, payment_secret, amt_msat, None, nodes[2].node.get_our_node_id()); let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]]; do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], path, payment_preimage)); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(nodes[0], payment_preimage); } #[test] @@ -1810,7 +1812,7 @@ fn test_manager_persisted_post_outbound_edge_holding_cell() { // Claim the a->b->c payment on node_c. let path: &[&[_]] = &[&[&nodes[1], &nodes[2]]]; do_claim_payment_along_route(ClaimAlongRouteArgs::new(&nodes[0], path, payment_preimage)); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); // Claim the c->b payment on node_b. nodes[1].node.claim_funds(payment_preimage_2); @@ -1819,7 +1821,7 @@ fn test_manager_persisted_post_outbound_edge_holding_cell() { let mut update = get_htlc_update_msgs(&nodes[1], &nodes[2].node.get_our_node_id()); nodes[2].node.handle_update_fulfill_htlc(nodes[1].node.get_our_node_id(), update.update_fulfill_htlcs.remove(0)); do_commitment_signed_dance(&nodes[2], &nodes[1], &update.commitment_signed, false, false); - expect_payment_sent(&nodes[2], payment_preimage_2, None, true, true); + expect_payment_sent!(&nodes[2], payment_preimage_2); } #[test] @@ -2231,7 +2233,7 @@ fn outbound_removed_holding_cell_resolved_no_double_forward() { reconnect_nodes(reconnect_args); // nodes[0] should now have received the fulfill and generate PaymentSent. - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } #[test] @@ -2335,7 +2337,7 @@ fn test_reload_node_with_preimage_in_monitor_claims_htlc() { reconnect_nodes(reconnect_args); // nodes[0] should now have received the fulfill and generate PaymentSent. - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } #[test] @@ -2596,5 +2598,5 @@ fn test_reload_with_mpp_claims_on_same_channel() { reconnect_nodes(reconnect_args); // nodes[0] should now have received both fulfills and generate PaymentSent. - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } diff --git a/lightning/src/ln/shutdown_tests.rs b/lightning/src/ln/shutdown_tests.rs index d70b240e4e4..5363b9b3b82 100644 --- a/lightning/src/ln/shutdown_tests.rs +++ b/lightning/src/ln/shutdown_tests.rs @@ -1864,7 +1864,10 @@ fn test_pending_htlcs_arent_lost_on_mon_delay() { let chanmon_cfgs = create_chanmon_cfgs(3); let node_cfgs = create_node_cfgs(3, &chanmon_cfgs); - let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]); + let mut cfg = test_default_channel_config(); + // When persistent_monitor_events is enabled, monitor updates will never be blocked. + cfg.override_persistent_monitor_events = Some(false); + let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, Some(cfg), None]); let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs); let node_a_id = nodes[0].node.get_our_node_id(); From 0deb66ffd89f354c9ace0e9876d5ba23cf83efde Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 16:27:30 -0400 Subject: [PATCH 18/20] Filter claims from get_onchain_failed_htlcs return value This isn't a bug at the moment because a claim in this situation would already be filtered out due to its inclusion in htlcs_resolved_to_user. However, when we stop issuing ReleasePaymentComplete monitor updates for claims in upcoming commits, HTLC claims will no longer be in htlcs_resolved_to_user when get_onchain_failed_htlcs checks. --- lightning/src/chain/channelmonitor.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 5d7968c5db5..a8a22bb33f8 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -3318,7 +3318,11 @@ impl ChannelMonitor { // The HTLC was not included in the confirmed commitment transaction, // which has now reached ANTI_REORG_DELAY confirmations and thus the // HTLC has been failed. - res.insert(source.clone(), candidate_htlc.payment_hash); + // However, if we have the preimage, the HTLC was fulfilled off-chain + // and should not be reported as failed. + if !us.counterparty_fulfilled_htlcs.contains_key(&htlc_id) { + res.insert(source.clone(), candidate_htlc.payment_hash); + } } } }; From 00ad38dd2a1fc7391458f4ce49c7078faede920f Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 3 Apr 2026 16:45:45 -0400 Subject: [PATCH 19/20] Persistent monitor events for onchain outbound claims Currently, the resolution of HTLCs (and decisions on when HTLCs can be forwarded) is the responsibility of Channel objects (a part of ChannelManager) until the channel is closed, and then the ChannelMonitor thereafter. This leads to some complexity around race conditions for HTLCs right around channel closure. Additionally, there is lots of complexity reconstructing the state of all HTLCs in the ChannelManager deserialization/loading logic. Instead, we want to do all resolution in ChannelMonitors (in response to ChannelMonitorUpdates) and pass them back to ChannelManager in the form of MonitorEvents (similar to how HTLCs are resolved after channels are closed). In order to have reliable resolution, we'll need to keep MonitorEvents around in the ChannelMonitor until the ChannelManager has finished processing them. This will simplify things - on restart instead of examining the set of HTLCs in monitors we can simply replay all the pending MonitorEvents. In recent work, we added support for keeping monitor events around until they are explicitly acked by the ChannelManager, but would always ack monitor events immediately, which preserved the previous behavior and didn't break any tests. Here we start acking monitor events for on-chain HTLC claims when the user processes the PaymentSent event, if the persistent_monitor_events feature is enabled. This allows us to stop issuing ReleasePaymentComplete monitor updates for onchain payment claims, because the purpose of that behavior is to ensure we won't keep repeatedly issuing PaymentSent events, and the monitor event and acking it on PaymentSent processing now serves that purpose instead. --- .../tests/lsps2_integration_tests.rs | 4 +- lightning/src/ln/chanmon_update_fail_tests.rs | 2 +- lightning/src/ln/channelmanager.rs | 37 +++++++++++------ lightning/src/ln/functional_tests.rs | 5 ++- lightning/src/ln/monitor_tests.rs | 26 ++++++------ lightning/src/ln/payment_tests.rs | 40 +++++++++++-------- lightning/src/ln/reorg_tests.rs | 7 +++- lightning/src/ln/splicing_tests.rs | 7 +++- 8 files changed, 80 insertions(+), 48 deletions(-) diff --git a/lightning-liquidity/tests/lsps2_integration_tests.rs b/lightning-liquidity/tests/lsps2_integration_tests.rs index 92e6b33ebb6..9e4d452e1a0 100644 --- a/lightning-liquidity/tests/lsps2_integration_tests.rs +++ b/lightning-liquidity/tests/lsps2_integration_tests.rs @@ -8,7 +8,6 @@ use common::{ }; use lightning::events::{ClosureReason, Event}; -use lightning::get_event_msg; use lightning::ln::channelmanager::{ OptionalBolt11PaymentParams, PaymentId, TrustedChannelFeatures, }; @@ -17,6 +16,7 @@ use lightning::ln::msgs::BaseMessageHandler; use lightning::ln::msgs::ChannelMessageHandler; use lightning::ln::msgs::MessageSendEvent; use lightning::ln::types::ChannelId; +use lightning::{expect_payment_sent, get_event_msg}; use lightning_liquidity::events::LiquidityEvent; use lightning_liquidity::lsps0::ser::LSPSDateTime; @@ -1340,7 +1340,7 @@ fn client_trusts_lsp_end_to_end_test() { let broadcasted = service_node.inner.tx_broadcaster.txn_broadcasted.lock().unwrap(); assert!(broadcasted.iter().any(|b| b.compute_txid() == funding_tx.compute_txid())); - expect_payment_sent(&payer_node, preimage.unwrap(), Some(total_fee_msat), true, true); + expect_payment_sent!(&payer_node, preimage.unwrap(), total_fee_msat); } fn execute_lsps2_dance( diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs index 4ef20358848..ba4cb17c985 100644 --- a/lightning/src/ln/chanmon_update_fail_tests.rs +++ b/lightning/src/ln/chanmon_update_fail_tests.rs @@ -3954,7 +3954,7 @@ fn do_test_durable_preimages_on_closed_channel( mine_transactions(&nodes[0], &[&as_closing_tx[0], bs_preimage_tx]); check_closed_broadcast(&nodes[0], 1, false); - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); if close_chans_before_reload && !hold_post_reload_mon_update { // For close_chans_before_reload with hold=false, the deferred completions diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index f18e207893c..7bb70a76a54 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -10180,7 +10180,18 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ "We don't support claim_htlc claims during startup - monitors may not be available yet"); debug_assert_eq!(next_channel_counterparty_node_id, path.hops[0].pubkey); - let mut ev_completion_action = if from_onchain { + let mut ev_completion_action = if self.persistent_monitor_events { + // If persistent monitor events is enabled: + // * If the channel is on-chain, we don't need to use ReleasePaymentComplete like we do + // below because we will stop hearing about this payment after the relevant monitor event + // is acked + // * If the channel is open, we don't need to block the preimage-removing monitor update + // like we do below because we will keep hearing about the preimage until we explicitly + // ack the monitor event for this payment + monitor_event_id.map(|event_id| EventCompletionAction::AckMonitorEvent { + event_id: MonitorEventSource { channel_id: next_channel_id, event_id }, + }) + } else if from_onchain { let release = PaymentCompleteUpdate { counterparty_node_id: next_channel_counterparty_node_id, channel_funding_outpoint: next_channel_outpoint, @@ -10189,17 +10200,11 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ }; Some(EventCompletionAction::ReleasePaymentCompleteChannelMonitorUpdate(release)) } else { - if self.persistent_monitor_events { - monitor_event_id.map(|event_id| EventCompletionAction::AckMonitorEvent { - event_id: MonitorEventSource { channel_id: next_channel_id, event_id }, - }) - } else { - Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate { - channel_funding_outpoint: Some(next_channel_outpoint), - channel_id: next_channel_id, - counterparty_node_id: path.hops[0].pubkey, - }) - } + Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate { + channel_funding_outpoint: Some(next_channel_outpoint), + channel_id: next_channel_id, + counterparty_node_id: path.hops[0].pubkey, + }) }; let logger = WithContext::for_payment( &self.logger, @@ -13810,7 +13815,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ Some(event_id), ); } - if from_onchain | !we_are_sender { + if !we_are_sender { self.chain_monitor.ack_monitor_event(monitor_event_source); } } else { @@ -20048,6 +20053,12 @@ impl< .. } => { if let Some(preimage) = preimage_opt { + if persistent_monitor_events { + // If persistent_monitor_events is enabled, then the monitor will keep + // providing us with a monitor event for this claim until the ChannelManager + // explicitly marks it as resolved, no need to explicitly handle it here. + continue; + } let pending_events = Mutex::new(pending_events_read); let update = PaymentCompleteUpdate { counterparty_node_id: monitor.get_counterparty_node_id(), diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs index 12b6beb0621..394c81f3e45 100644 --- a/lightning/src/ln/functional_tests.rs +++ b/lightning/src/ln/functional_tests.rs @@ -1632,7 +1632,10 @@ pub fn test_htlc_on_chain_success() { check_closed_broadcast(&nodes[0], 1, true); check_added_monitors(&nodes[0], 1); let events = nodes[0].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[0], 2); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 2 }, + ); assert_eq!(events.len(), 5); let mut first_claimed = false; for event in events { diff --git a/lightning/src/ln/monitor_tests.rs b/lightning/src/ln/monitor_tests.rs index 043ecddf7b5..5dca3c4a582 100644 --- a/lightning/src/ln/monitor_tests.rs +++ b/lightning/src/ln/monitor_tests.rs @@ -274,7 +274,7 @@ fn archive_fully_resolved_monitors() { // Finally, we process the pending `MonitorEvent` from nodes[0], allowing the `ChannelMonitor` // to be archived `ARCHIVAL_DELAY_BLOCKS` blocks later. - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors(); assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 1); connect_blocks(&nodes[0], ARCHIVAL_DELAY_BLOCKS - 1); @@ -747,9 +747,9 @@ fn do_test_claim_value_force_close(keyed_anchors: bool, p2a_anchor: bool, prev_c mine_transaction(&nodes[0], &b_broadcast_txn[0]); if prev_commitment_tx { expect_payment_path_successful!(nodes[0]); - check_added_monitors(&nodes[0], 1); + check_added_monitors(&nodes[0], if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 1 }); } else { - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } assert_eq!(sorted_vec(vec![sent_htlc_balance.clone(), sent_htlc_timeout_balance.clone()]), sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(chan_id).unwrap().get_claimable_balances())); @@ -1015,7 +1015,7 @@ fn do_test_balances_on_local_commitment_htlcs(keyed_anchors: bool, p2a_anchor: b // Now confirm nodes[1]'s HTLC claim, giving nodes[0] the preimage. Note that the "maybe // claimable" balance remains until we see ANTI_REORG_DELAY blocks. mine_transaction(&nodes[0], &bs_htlc_claim_txn[0]); - expect_payment_sent(&nodes[0], payment_preimage_2, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage_2); assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations { amount_satoshis: 1_000_000 - 10_000 - 20_000 - commitment_tx_fee - anchor_outputs_value, confirmation_height: node_a_commitment_claimable, @@ -2097,7 +2097,7 @@ fn do_test_revoked_counterparty_aggregated_claims(keyed_anchors: bool, p2a_ancho as_revoked_txn[1].clone() }; mine_transaction(&nodes[1], &htlc_success_claim); - expect_payment_sent(&nodes[1], claimed_payment_preimage, None, true, true); + expect_payment_sent!(&nodes[1], claimed_payment_preimage); let mut claim_txn_2 = nodes[1].tx_broadcaster.txn_broadcast(); // Once B sees the HTLC-Success transaction it splits its claim transaction into two, though in @@ -3547,9 +3547,13 @@ fn do_test_lost_preimage_monitor_events(on_counterparty_tx: bool, p2a_anchor: bo // After the background events are processed in `get_and_clear_pending_events`, above, node B // will create the requisite `ChannelMontiorUpdate` for claiming the forwarded payment back. // The HTLC, however, is added to the holding cell for replay after the peer connects, below. - // It will also apply a `ChannelMonitorUpdate` to let the `ChannelMonitor` know that the - // payment can now be forgotten as the `PaymentSent` event was handled. - check_added_monitors(&nodes[1], 2); + if nodes[1].node.test_persistent_monitor_events_enabled() { + check_added_monitors(&nodes[1], 1); + } else { + // It will also apply a `ChannelMonitorUpdate` to let the `ChannelMonitor` know that the + // payment can now be forgotten as the `PaymentSent` event was handled. + check_added_monitors(&nodes[1], 2); + } nodes[0].node.peer_disconnected(nodes[1].node.get_our_node_id()); @@ -3943,8 +3947,7 @@ fn test_ladder_preimage_htlc_claims() { connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1); connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1); - expect_payment_sent(&nodes[0], payment_preimage1, None, true, false); - check_added_monitors(&nodes[0], 1); + expect_payment_sent!(&nodes[0], payment_preimage1); nodes[1].node.claim_funds(payment_preimage2); expect_payment_claimed!(&nodes[1], payment_hash2, 1_000_000); @@ -3966,6 +3969,5 @@ fn test_ladder_preimage_htlc_claims() { connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1); connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1); - expect_payment_sent(&nodes[0], payment_preimage2, None, true, false); - check_added_monitors(&nodes[0], 1); + expect_payment_sent!(&nodes[0], payment_preimage2); } diff --git a/lightning/src/ln/payment_tests.rs b/lightning/src/ln/payment_tests.rs index 59725f62bc1..cb14a6d9f8a 100644 --- a/lightning/src/ln/payment_tests.rs +++ b/lightning/src/ln/payment_tests.rs @@ -992,7 +992,7 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) { assert_eq!(txn[0].compute_txid(), as_commitment_tx.compute_txid()); } mine_transaction(&nodes[0], &bs_htlc_claim_txn); - expect_payment_sent(&nodes[0], payment_preimage_1, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage_1); connect_blocks(&nodes[0], TEST_FINAL_CLTV * 4 + 20); let (first_htlc_timeout_tx, second_htlc_timeout_tx) = { let mut txn = nodes[0].tx_broadcaster.unique_txn_broadcast(); @@ -1403,7 +1403,7 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload( let conditions = PaymentFailedConditions::new().from_mon_update(); expect_payment_failed_conditions(&nodes[0], payment_hash, false, conditions); } else { - expect_payment_sent(&nodes[0], payment_preimage, None, true, true); + expect_payment_sent!(&nodes[0], payment_preimage); } // Note that if we persist the monitor before processing the events, above, we'll always get // them replayed on restart no matter what @@ -1441,18 +1441,22 @@ fn do_test_dup_htlc_onchain_doesnt_fail_on_reload( } else { expect_payment_sent(&nodes[0], payment_preimage, None, true, false); } - if persist_manager_post_event { - // After reload, the ChannelManager identified the failed payment and queued up the - // PaymentSent (or not, if `persist_manager_post_event` resulted in us detecting we - // already did that) and corresponding ChannelMonitorUpdate to mark the payment - // handled, but while processing the pending `MonitorEvent`s (which were not processed - // before the monitor was persisted) we will end up with a duplicate - // ChannelMonitorUpdate. - check_added_monitors(&nodes[0], 2); + if !nodes[0].node.test_persistent_monitor_events_enabled() { + if persist_manager_post_event { + // After reload, the ChannelManager identified the failed payment and queued up the + // PaymentSent (or not, if `persist_manager_post_event` resulted in us detecting we + // already did that) and corresponding ChannelMonitorUpdate to mark the payment + // handled, but while processing the pending `MonitorEvent`s (which were not processed + // before the monitor was persisted) we will end up with a duplicate + // ChannelMonitorUpdate. + check_added_monitors(&nodes[0], 2); + } else { + // ...unless we got the PaymentSent event, in which case we have de-duplication logic + // preventing a redundant monitor event. + check_added_monitors(&nodes[0], 1); + } } else { - // ...unless we got the PaymentSent event, in which case we have de-duplication logic - // preventing a redundant monitor event. - check_added_monitors(&nodes[0], 1); + check_added_monitors(&nodes[0], 0); } } @@ -4255,10 +4259,14 @@ fn do_no_missing_sent_on_reload(persist_manager_with_payment: bool, at_midpoint: check_added_monitors(&nodes[0], 0); nodes[0].node.test_process_background_events(); check_added_monitors(&nodes[0], 1); - // Then once we process the PaymentSent event we'll apply a monitor update to remove the - // pending payment from being re-hydrated on the next startup. let events = nodes[0].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[0], 1); + if nodes[0].node.test_persistent_monitor_events_enabled() { + check_added_monitors(&nodes[0], 0); + } else { + // Once we process the PaymentSent event we'll apply a monitor update to remove the + // pending payment from being re-hydrated on the next startup. + check_added_monitors(&nodes[0], 1); + } assert_eq!(events.len(), 3, "{events:?}"); if let Event::ChannelClosed { reason: ClosureReason::OutdatedChannelManager, .. } = events[0] { } else { diff --git a/lightning/src/ln/reorg_tests.rs b/lightning/src/ln/reorg_tests.rs index 5b5160148d7..b99c5aa936a 100644 --- a/lightning/src/ln/reorg_tests.rs +++ b/lightning/src/ln/reorg_tests.rs @@ -254,7 +254,7 @@ fn test_counterparty_revoked_reorg() { // Connect the HTLC claim transaction for HTLC 3 mine_transaction(&nodes[1], &unrevoked_local_txn[2]); - expect_payment_sent(&nodes[1], payment_preimage_3, None, true, true); + expect_payment_sent!(&nodes[1], payment_preimage_3); assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); // Connect blocks to confirm the unrevoked commitment transaction @@ -1228,7 +1228,10 @@ fn do_test_split_htlc_expiry_tracking(use_third_htlc: bool, reorg_out: bool, p2a check_added_monitors(&nodes[0], 0); let sent_events = nodes[0].node.get_and_clear_pending_events(); - check_added_monitors(&nodes[0], 2); + check_added_monitors( + &nodes[0], + if nodes[0].node.test_persistent_monitor_events_enabled() { 0 } else { 2 }, + ); assert_eq!(sent_events.len(), 4, "{sent_events:?}"); let mut found_expected_events = [false, false, false, false]; for event in sent_events { diff --git a/lightning/src/ln/splicing_tests.rs b/lightning/src/ln/splicing_tests.rs index 35c72509d0b..278c2f76e08 100644 --- a/lightning/src/ln/splicing_tests.rs +++ b/lightning/src/ln/splicing_tests.rs @@ -2266,7 +2266,12 @@ fn do_test_splice_commitment_broadcast(splice_status: SpliceStatus, claim_htlcs: 2 ); } - check_added_monitors(&nodes[0], 2); // Two `ReleasePaymentComplete` monitor updates + let expected_mons = if nodes[0].node.test_persistent_monitor_events_enabled() && claim_htlcs { + 0 + } else { + 2 // Two `ReleasePaymentComplete` monitor updates + }; + check_added_monitors(&nodes[0], expected_mons); // When the splice never confirms and we see a commitment transaction broadcast and confirm for // the current funding instead, we should expect to see an `Event::DiscardFunding` for the From eb68e8affcf9bb65b2d0e3385f873d80ad36835a Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Fri, 8 May 2026 15:44:15 -0400 Subject: [PATCH 20/20] Check for dangling monitor events in tests We want to make sure we don't leak any monitor events by leaving them unacked forever. --- lightning/src/chain/channelmonitor.rs | 14 ++++++++++++-- lightning/src/ln/functional_test_utils.rs | 12 ++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index a8a22bb33f8..43378eda8ea 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -194,7 +194,7 @@ fn push_monitor_event( } /// An event to be processed by the ChannelManager. -#[derive(Clone, PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq, Debug)] pub enum MonitorEvent { /// A monitor event containing an HTLCUpdate. HTLCEvent(HTLCUpdate), @@ -256,7 +256,7 @@ impl_writeable_tlv_based_enum_upgradable_legacy!(MonitorEvent, /// Simple structure sent back by `chain::Watch` when an HTLC from a forward channel is detected on /// chain. Used to update the corresponding HTLC in the backward channel. Failing to pass the /// preimage claim backward will lead to loss of funds. -#[derive(Clone, PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq, Debug)] pub struct HTLCUpdate { pub(crate) payment_hash: PaymentHash, pub(crate) payment_preimage: Option, @@ -2272,6 +2272,16 @@ impl ChannelMonitor { self_inner.next_monitor_event_id = next_id; } + /// Used by test infra to check for monitor event leaks at the end of a test. + #[cfg(any(test, feature = "_test_utils"))] + pub fn drain_unacked_monitor_events(&self) -> Vec<(u64, MonitorEvent)> { + let mut inner = self.inner.lock().unwrap(); + let mut events = Vec::new(); + events.append(&mut inner.pending_monitor_events); + events.append(&mut inner.provided_monitor_events); + events + } + /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. /// /// For channels featuring anchor outputs, this method will also process [`BumpTransaction`] diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs index 01b32a4bf9c..d8d7ca921ff 100644 --- a/lightning/src/ln/functional_test_utils.rs +++ b/lightning/src/ln/functional_test_utils.rs @@ -801,6 +801,18 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> { panic!("Had excess RAA blockers on node {}: {:?}", self.logger.id, raa_blockers); } + for channel_id in self.chain_monitor.chain_monitor.list_monitors() { + let monitor = self.chain_monitor.chain_monitor.get_monitor(channel_id).unwrap(); + let unacked_monitor_events = monitor.drain_unacked_monitor_events(); + if !unacked_monitor_events.is_empty() { + panic!( + "Node {} channel {channel_id:?} had {} unacked monitor events at drop: {unacked_monitor_events:#?}", + unacked_monitor_events.len(), + self.logger.id + ); + } + } + // Check that if we serialize the network graph, we can deserialize it again. let network_graph = { let mut w = test_utils::TestVecWriter(Vec::new());