Persistent MonitorEvents

fuzz/src/chanmon_consistency.rs

@@ -41,6 +41,7 @@ use lightning::chain;
 use lightning::chain::chaininterface::{
 	BroadcasterInterface, ConfirmationTarget, FeeEstimator, TransactionType,
 };
+use lightning::chain::chainmonitor::MonitorEventSource;
 use lightning::chain::channelmonitor::{ChannelMonitor, MonitorEvent};
 use lightning::chain::transaction::OutPoint;
 use lightning::chain::{
@@ -363,9 +364,13 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 
 	fn release_pending_monitor_events(
 		&self,
-	) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> {
+	) -> Vec<(OutPoint, ChannelId, Vec<(u64, MonitorEvent)>, PublicKey)> {
 		return self.chain_monitor.release_pending_monitor_events();
 	}
+
+	fn ack_monitor_event(&self, source: MonitorEventSource) {
+		self.chain_monitor.ack_monitor_event(source);
+	}
 }
 
 struct KeyProvider {

lightning/src/chain/chainmonitor.rs

@@ -67,6 +67,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, {
+	(0, event_id, required),
+	(2, 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.
@@ -367,9 +382,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,
 
@@ -437,7 +449,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 },
@@ -656,7 +667,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()),
@@ -801,16 +811,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,14 +828,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();
 	}
 
@@ -1610,11 +1612,11 @@ 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 mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
+		let mut pending_monitor_events = Vec::new();
 		for monitor_state in self.monitors.read().unwrap().values() {
 			let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();
 			if monitor_events.len() > 0 {
@@ -1631,6 +1633,13 @@ 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);
+		}
+	}
 }
 
 impl<