Hold back HTLC mon events while updates in-progress

Recently, we began generating monitor events when HTLCs fail
off-chain. In upcoming commits, we'll begin relying on those events
to resolve forwarded HTLC fails on the inbound edge.

One requirement for that is that we can't fail an HTLC backwards from a monitor
event until the monitor update removing that HTLC on the outbound edge is
durably persisted. Otherwise, if we crash and lose that monitor update at the
wrong time, the outbound edge counterparty could theoretically change their
fail resolution to a claim and we'd have no recourse because we already failed
the HTLC back on the inbound edge.

As such, here we begin not surfacing HTLC-related monitor events to the
ChannelManager if there are in-flight monitor updates present, to prevent this
scenario.

Author: valentinewallace

lightning/src/chain/chainmonitor.rs

@@ -828,7 +828,10 @@ where
 	#[cfg(any(test, fuzzing))]
 	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 monitor_state = monitors.get(&channel_id).unwrap();
+		let monitor = &monitor_state.monitor;
+		let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
+		pending_monitor_updates.retain(|update_id| *update_id > monitor_update_id);
 		monitor.push_monitor_event(MonitorEvent::Completed {
 			funding_txo: monitor.get_funding_txo(),
 			channel_id,
@@ -1646,7 +1649,20 @@ where
 		let monitors = self.monitors.read().unwrap();
 		let mut pending_monitor_events = Vec::new();
 		for monitor_state in monitors.values() {
-			let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();
+			// Hold back HTLC monitor events for channels with in-flight updates. The monitor may have
+			// queued an event based on in-memory state from an as-yet-unpersisted update; surfacing it
+			// before persistence would let us act on (e.g. fail upstream) state that could be lost on a
+			// crash + reconnect. Other monitor events (e.g., channel close) aren't subject to those
+			// restrictions and can be released immediately.
+			let has_pending_updates = {
+				let pending_updates = monitor_state.pending_monitor_updates.lock().unwrap();
+				monitor_state.has_pending_updates(&pending_updates)
+			};
+			let monitor_events = if has_pending_updates {
+				monitor_state.monitor.get_and_clear_pending_non_htlc_monitor_events()
+			} else {
+				monitor_state.monitor.get_and_clear_pending_monitor_events()
+			};
 			if monitor_events.len() > 0 {
 				let monitor_funding_txo = monitor_state.monitor.get_funding_txo();
 				let monitor_channel_id = monitor_state.monitor.channel_id();

lightning/src/chain/channelmonitor.rs

@@ -2361,7 +2361,17 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	/// 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<(u64, MonitorEvent)> {
-		self.inner.lock().unwrap().get_and_clear_pending_monitor_events()
+		self.inner.lock().unwrap().get_and_clear_pending_monitor_events_filtered(|_| true)
+	}
+
+	/// Drains and returns pending monitor events except [`MonitorEvent::HTLCEvent`]s. Used by the
+	/// [`crate::chain::chainmonitor::ChainMonitor`] to avoid surfacing HTLC resolutions that arise
+	/// from a monitor update which has been propagated to the monitor in-memory but not yet durably
+	/// persisted.
+	pub(super) fn get_and_clear_pending_non_htlc_monitor_events(&self) -> Vec<(u64, MonitorEvent)> {
+		self.inner.lock().unwrap().get_and_clear_pending_monitor_events_filtered(|ev| {
+			!matches!(ev, MonitorEvent::HTLCEvent(_))
+		})
 	}
 
 	pub(super) fn push_monitor_event(&self, event: MonitorEvent) {
@@ -4937,17 +4947,27 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		})
 	}
 
-	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<(u64, MonitorEvent)> {
+	/// Drains and returns the pending monitor events for which `predicate` returns true. Events that
+	/// don't match the predicate stay in `pending_monitor_events` so they're eligible for release on
+	/// a later call.
+	fn get_and_clear_pending_monitor_events_filtered<F: FnMut(&MonitorEvent) -> bool>(
+		&mut self, mut predicate: F,
+	) -> Vec<(u64, MonitorEvent)> {
+		let mut released = Vec::new();
+		let mut retained = Vec::new();
+		// Note: we can use Vec::extract_if here once MSRV reaches 1.87
+		for entry in self.pending_monitor_events.drain(..) {
+			if predicate(&entry.1) {
+				released.push(entry);
+			} else {
+				retained.push(entry);
+			}
+		}
+		self.pending_monitor_events = retained;
 		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
+			self.provided_monitor_events.extend(released.iter().cloned());
 		}
+		released
 	}
 
 	/// Gets the set of events that are repeated regularly (e.g. those which RBF bump

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -5553,15 +5553,8 @@ fn native_async_persist() {
 
 	persist_futures.poll_futures();
 	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());
-	}
+	// HTLC resolution monitor events are held back while monitor updates are in-progress.
+	assert!(events.is_empty());
 
 	let pending_writes = kv_store.list_pending_async_writes(
 		CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
@@ -5593,12 +5586,29 @@ fn native_async_persist() {
 	);
 	persist_futures.poll_futures();
 	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] {
-		assert_eq!(*monitor_update_id, 4);
+	if persistent_monitor_events {
+		let mon_evs: Vec<_> = completed_persist.iter().flat_map(|(_, _, evs, _)| evs).collect();
+		assert_eq!(mon_evs.len(), 2);
+		// We previously held back an HTLC event because there was an in-flight update in progress, but
+		// now that that's completed the event will be surfaced.
+		match mon_evs[0].1 {
+			MonitorEvent::HTLCEvent(..) => {},
+			_ => panic!(),
+		}
+		match mon_evs[1].1 {
+			MonitorEvent::Completed { monitor_update_id, .. } => {
+				assert_eq!(monitor_update_id, 4);
+			},
+			_ => panic!(),
+		}
 	} else {
-		panic!();
+		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] {
+			assert_eq!(*monitor_update_id, 4);
+		} else {
+			panic!();
+		}
 	}
 }