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.

Author: valentinewallace

lightning/src/chain/channelmonitor.rs

@@ -1303,6 +1303,11 @@ pub(crate) struct ChannelMonitorImpl<Signer: EcdsaChannelSigner> {
 	// 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<Signer: EcdsaChannelSigner, W: Writer>(
 	// 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<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 			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,20 +2232,32 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 	/// 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`.
 	/// 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<Signer>) {
-		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<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		);
 	}
 
+	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<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				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<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 			// 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<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 						}), &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,

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 {

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);

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<bool>,
 }
 
 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,
 		})
 	}
 }