f Trigger ChainMonitor's event_notifier on failure

Author: tnull

lightning/src/chain/chainmonitor.rs

@@ -551,8 +551,13 @@ where C::Target: chain::Filter,
 		let mons_to_process = self.monitors.read().unwrap().keys().cloned().collect::<Vec<_>>();
 		for funding_txo in mons_to_process {
 			let mut ev;
-			super::channelmonitor::process_events_body!(
-				self.monitors.read().unwrap().get(&funding_txo).map(|m| &m.monitor), ev, handler(ev).await);
+			match super::channelmonitor::process_events_body!(
+				self.monitors.read().unwrap().get(&funding_txo).map(|m| &m.monitor), ev, handler(ev).await) {
+				Ok(()) => {},
+				Err(ReplayEvent ()) => {
+					self.event_notifier.notify();
+				}
+			}
 		}
 	}
 
@@ -879,7 +884,12 @@ impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref,
 	/// [`BumpTransaction`]: events::Event::BumpTransaction
 	fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
 		for monitor_state in self.monitors.read().unwrap().values() {
-			monitor_state.monitor.process_pending_events(&handler);
+			match monitor_state.monitor.process_pending_events(&handler) {
+				Ok(()) => {},
+				Err(ReplayEvent ()) => {
+					self.event_notifier.notify();
+				}
+			}
 		}
 	}
 }

lightning/src/chain/channelmonitor.rs

@@ -1159,28 +1159,28 @@ impl<Signer: EcdsaChannelSigner> Writeable for ChannelMonitorImpl<Signer> {
 macro_rules! _process_events_body {
 	($self_opt: expr, $event_to_handle: expr, $handle_event: expr) => {
 		loop {
+			let mut handling_res = Ok(());
 			let (pending_events, repeated_events);
 			if let Some(us) = $self_opt {
 				let mut inner = us.inner.lock().unwrap();
 				if inner.is_processing_pending_events {
-					break;
+					break handling_res;
 				}
 				inner.is_processing_pending_events = true;
 
 				pending_events = inner.pending_events.clone();
 				repeated_events = inner.get_repeated_events();
-			} else { break; }
+			} else { break handling_res; }
 
 			let mut num_handled_events = 0;
-			let mut handling_failed = false;
 			for event in pending_events.into_iter() {
 				$event_to_handle = event;
 				match $handle_event {
 					Ok(()) => num_handled_events += 1,
-					Err(_) => {
+					Err(e) => {
 						// If we encounter an error we stop handling events and make sure to replay
 						// any unhandled events on the next invocation.
-						handling_failed = true;
+						handling_res = Err(e);
 						break;
 					}
 				}
@@ -1197,13 +1197,13 @@ macro_rules! _process_events_body {
 				let mut inner = us.inner.lock().unwrap();
 				inner.pending_events.drain(..num_handled_events);
 				inner.is_processing_pending_events = false;
-				if !handling_failed && !inner.pending_events.is_empty() {
+				if handling_res.is_ok() && !inner.pending_events.is_empty() {
 					// If there's more events to process and we didn't fail so far, go ahead and do
 					// so.
 					continue;
 				}
 			}
-			break;
+			break handling_res;
 		}
 	}
 }
@@ -1515,21 +1515,23 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 	/// An [`EventHandler`] may safely call back to the provider, though this shouldn't be needed in
 	/// order to handle these events.
 	///
+	/// Will return a [`ReplayEvent`] error if event handling failed and should eventually be retried.
+	///
 	/// [`SpendableOutputs`]: crate::events::Event::SpendableOutputs
 	/// [`BumpTransaction`]: crate::events::Event::BumpTransaction
-	pub fn process_pending_events<H: Deref>(&self, handler: &H) where H::Target: EventHandler {
+	pub fn process_pending_events<H: Deref>(&self, handler: &H) -> Result<(), ReplayEvent> where H::Target: EventHandler {
 		let mut ev;
-		process_events_body!(Some(self), ev, handler.handle_event(ev));
+		process_events_body!(Some(self), ev, handler.handle_event(ev))
 	}
 
 	/// Processes any events asynchronously.
 	///
 	/// See [`Self::process_pending_events`] for more information.
 	pub async fn process_pending_events_async<Future: core::future::Future<Output = Result<(), ReplayEvent>>, H: Fn(Event) -> Future>(
 		&self, handler: &H
-	) {
+	) -> Result<(), ReplayEvent> {
 		let mut ev;
-		process_events_body!(Some(self), ev, { handler(ev).await });
+		process_events_body!(Some(self), ev, { handler(ev).await })
 	}
 
 	#[cfg(test)]