Resolve forward HTLC fails via MonitorEvents

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(

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_ser_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<ChannelSigner: EcdsaChannelSigner> {
 	/// A new monitor to insert and persist.
@@ -366,9 +381,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<Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)>>,
 	/// The best block height seen, used as a proxy for the passage of time.
 	highest_chain_height: AtomicUsize,
 
@@ -436,7 +448,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 +668,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 +812,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(())
@@ -823,15 +828,15 @@ 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 counterparty_node_id = monitor.get_counterparty_node_id();
-		let funding_txo = monitor.get_funding_txo();
-		self.pending_monitor_events.lock().unwrap().push((
-			funding_txo,
+		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,
-			vec![MonitorEvent::Completed { funding_txo, channel_id, monitor_update_id }],
-			counterparty_node_id,
-		));
+			monitor_update_id,
+		});
 		self.event_notifier.notify();
 	}
 
@@ -1266,21 +1271,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)",
@@ -1645,14 +1642,27 @@ where
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, 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);
 		}
 		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();
@@ -1665,12 +1675,17 @@ 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
 	}
+
+	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<