Add helper to push monitor events

Cleans up the next commit

Author: valentinewallace

lightning/src/chain/chainmonitor.rs

@@ -366,9 +366,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 +433,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 +653,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 +797,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(())
@@ -824,14 +814,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();
 	}
 
@@ -1266,21 +1253,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)",
@@ -1665,10 +1644,6 @@ 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
 	}
 }

lightning/src/chain/channelmonitor.rs

@@ -183,6 +183,10 @@ impl Readable for ChannelMonitorUpdate {
 	}
 }
 
+fn push_monitor_event(pending_monitor_events: &mut Vec<MonitorEvent>, event: MonitorEvent) {
+	pending_monitor_events.push(event);
+}
+
 /// An event to be processed by the ChannelManager.
 #[derive(Clone, PartialEq, Eq)]
 pub enum MonitorEvent {
@@ -226,8 +230,6 @@ pub enum MonitorEvent {
 	},
 }
 impl_writeable_tlv_based_enum_upgradable_legacy!(MonitorEvent,
-	// Note that Completed is currently never serialized to disk as it is generated only in
-	// ChainMonitor.
 	(0, Completed) => {
 		(0, funding_txo, required),
 		(2, monitor_update_id, required),
@@ -2166,6 +2168,10 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		self.inner.lock().unwrap().get_and_clear_pending_monitor_events()
 	}
 
+	pub(super) fn push_monitor_event(&self, event: MonitorEvent) {
+		self.inner.lock().unwrap().push_monitor_event(event);
+	}
+
 	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
 	///
 	/// For channels featuring anchor outputs, this method will also process [`BumpTransaction`]
@@ -3891,7 +3897,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 				outpoint: funding_outpoint,
 				channel_id: self.channel_id,
 			};
-			self.pending_monitor_events.push(event);
+			push_monitor_event(&mut self.pending_monitor_events, event);
 		}
 
 		// Although we aren't signing the transaction directly here, the transaction will be signed
@@ -4552,6 +4558,10 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		&self.outputs_to_watch
 	}
 
+	fn push_monitor_event(&mut self, event: MonitorEvent) {
+		push_monitor_event(&mut self.pending_monitor_events, event);
+	}
+
 	fn get_and_clear_pending_monitor_events(&mut self) -> Vec<MonitorEvent> {
 		let mut ret = Vec::new();
 		mem::swap(&mut ret, &mut self.pending_monitor_events);
@@ -5611,7 +5621,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					);
 					log_info!(logger, "Channel closed by funding output spend in txid {txid}");
 					if !self.funding_spend_seen {
-						self.pending_monitor_events.push(MonitorEvent::CommitmentTxConfirmed(()));
+						self.push_monitor_event(MonitorEvent::CommitmentTxConfirmed(()));
 					}
 					self.funding_spend_seen = true;
 
@@ -5786,7 +5796,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 
 					log_debug!(logger, "HTLC {} failure update in {} has got enough confirmations to be passed upstream",
 						&payment_hash, entry.txid);
-					self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					self.push_monitor_event(MonitorEvent::HTLCEvent(HTLCUpdate {
 						payment_hash,
 						payment_preimage: None,
 						source,
@@ -5896,7 +5906,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 					log_error!(logger, "Failing back HTLC {} upstream to preserve the \
 						channel as the forward HTLC hasn't resolved and our backward HTLC \
 						expires soon at {}", log_bytes!(htlc.payment_hash.0), inbound_htlc_expiry);
-					self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate {
 						source: source.clone(),
 						payment_preimage: None,
 						payment_hash: htlc.payment_hash,
@@ -6313,7 +6323,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							},
 						});
 						self.counterparty_fulfilled_htlcs.insert(SentHTLCId::from_source(&source), payment_preimage);
-						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+						push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,
 							payment_preimage: Some(payment_preimage),
 							payment_hash,
@@ -6337,7 +6347,7 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							},
 						});
 						self.counterparty_fulfilled_htlcs.insert(SentHTLCId::from_source(&source), payment_preimage);
-						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+						push_monitor_event(&mut self.pending_monitor_events, MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,
 							payment_preimage: Some(payment_preimage),
 							payment_hash,