Persistent MonitorEvents

fuzz/src/chanmon_consistency.rs

@@ -362,9 +362,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, channel_id: ChannelId, event_id: u64) {
+		self.chain_monitor.ack_monitor_event(channel_id, event_id);
+	}
 }
 
 struct KeyProvider {

lightning/src/chain/chainmonitor.rs

@@ -356,9 +356,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,
 
@@ -407,7 +404,7 @@ 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 },
@@ -613,7 +610,7 @@ 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()),
@@ -755,16 +752,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(())
@@ -777,14 +769,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();
 	}
 
@@ -1425,11 +1414,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 {
@@ -1446,6 +1435,13 @@ where
 		}
 		pending_monitor_events
 	}
+
+	fn ack_monitor_event(&self, channel_id: ChannelId, event_id: u64) {
+		let monitors = self.monitors.read().unwrap();
+		if let Some(monitor_state) = monitors.get(&channel_id) {
+			monitor_state.monitor.ack_monitor_event(event_id);
+		}
+	}
 }
 
 impl<