Support persistent monitor events

Monitor events will keep being provided on restart to the ChannelManager
until they are explicitly ACK'd. This new code will run randomly in tests,
to ensure we still support the old paths.

XXX

Author: valentinewallace

lightning/src/chain/channelmonitor.rs

@@ -2191,7 +2191,24 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 
 	/// Removes a [`MonitorEvent`] by its event ID, acknowledging that it has been processed.
 	pub(super) fn ack_monitor_event(&self, event_id: u64) {
-		self.inner.lock().unwrap().pending_monitor_events.retain(|(id, _)| *id != event_id);
+		let inner = &mut *self.inner.lock().unwrap();
+		inner.pending_monitor_events.retain(|(id, _)| *id != event_id);
+		inner.provided_monitor_event_ids.remove(&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 the in-memory `provided_monitor_event_ids` 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_provided_monitor_event_ids(&self, other: &ChannelMonitor<Signer>) {
+		let ids = other.inner.lock().unwrap().provided_monitor_event_ids.clone();
+		self.inner.lock().unwrap().provided_monitor_event_ids = ids;
 	}
 
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
@@ -4418,9 +4435,22 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 	}
 
 	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 ret: Vec<_> = self
+				.pending_monitor_events
+				.iter()
+				.filter(|(id, _)| !self.provided_monitor_event_ids.contains(id))
+				.cloned()
+				.collect();
+			for (id, _) in &ret {
+				self.provided_monitor_event_ids.insert(*id);
+			}
+			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

lightning/src/ln/channelmanager.rs

@@ -3630,6 +3630,8 @@ impl<
 			our_network_pubkey, current_timestamp, expanded_inbound_key,
 			node_signer.get_receive_auth_key(), secp_ctx.clone(), message_router, logger.clone(),
 		);
+		#[cfg(test)]
+		let override_persistent_monitor_events = config.override_persistent_monitor_events;
 
 		ChannelManager {
 			config: RwLock::new(config),
@@ -3686,7 +3688,27 @@ impl<
 
 			logger,
 
-			persistent_monitor_events: false,
+			persistent_monitor_events: {
+				#[cfg(not(test))]
+				{ false }
+				#[cfg(test)]
+				{
+					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
+							},
+						}
+					})
+				}
+			},
 
 			#[cfg(feature = "_test_utils")]
 			testing_dnssec_proof_offer_resolution_override: Mutex::new(new_hash_map()),
@@ -11570,6 +11592,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
@@ -11740,6 +11763,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(|()| {
@@ -12655,6 +12679,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

@@ -3591,6 +3591,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

@@ -1103,6 +1103,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(test)]
+	pub override_persistent_monitor_events: Option<bool>,
 }
 
 impl Default for UserConfig {
@@ -1119,6 +1123,8 @@ impl Default for UserConfig {
 			enable_htlc_hold: false,
 			hold_outbound_htlcs_at_next_hop: false,
 			reject_inbound_splices: true,
+			#[cfg(test)]
+			override_persistent_monitor_events: None,
 		}
 	}
 }

lightning/src/util/test_utils.rs

@@ -606,6 +606,7 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		)
 		.unwrap()
 		.1;
+		new_monitor.copy_provided_monitor_event_ids(&monitor);
 		assert!(new_monitor == monitor);
 		self.latest_monitor_update_id
 			.lock()
@@ -662,6 +663,9 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		)
 		.unwrap()
 		.1;
+		// The deserialized monitor won't have provided_monitor_event_ids (it's in-memory only),
+		// so copy it from the live monitor before comparing.
+		new_monitor.copy_provided_monitor_event_ids(&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);