Make event handling fallible

lightning-background-processor/src/lib.rs

@@ -26,6 +26,8 @@ use lightning::chain::chainmonitor::{ChainMonitor, Persist};
 use lightning::events::EventHandler;
 #[cfg(feature = "std")]
 use lightning::events::EventsProvider;
+#[cfg(feature = "futures")]
+use lightning::events::ReplayEvent;
 use lightning::events::{Event, PathFailure};
 
 use lightning::ln::channelmanager::AChannelManager;
@@ -583,6 +585,7 @@ use futures_util::{dummy_waker, Selector, SelectorOutput};
 /// could setup `process_events_async` like this:
 /// ```
 /// # use lightning::io;
+/// # use lightning::events::ReplayEvent;
 /// # use std::sync::{Arc, RwLock};
 /// # use std::sync::atomic::{AtomicBool, Ordering};
 /// # use std::time::SystemTime;
@@ -600,7 +603,7 @@ use futures_util::{dummy_waker, Selector, SelectorOutput};
 /// # }
 /// # struct EventHandler {}
 /// # impl EventHandler {
-/// #     async fn handle_event(&self, _: lightning::events::Event) {}
+/// #     async fn handle_event(&self, _: lightning::events::Event) -> Result<(), ReplayEvent> { Ok(()) }
 /// # }
 /// # #[derive(Eq, PartialEq, Clone, Hash)]
 /// # struct SocketDescriptor {}
@@ -698,7 +701,7 @@ pub async fn process_events_async<
 	G: 'static + Deref<Target = NetworkGraph<L>> + Send + Sync,
 	L: 'static + Deref + Send + Sync,
 	P: 'static + Deref + Send + Sync,
-	EventHandlerFuture: core::future::Future<Output = ()>,
+	EventHandlerFuture: core::future::Future<Output = Result<(), ReplayEvent>>,
 	EventHandler: Fn(Event) -> EventHandlerFuture,
 	PS: 'static + Deref + Send,
 	M: 'static
@@ -751,12 +754,16 @@ where
 					if update_scorer(scorer, &event, duration_since_epoch) {
 						log_trace!(logger, "Persisting scorer after update");
 						if let Err(e) = persister.persist_scorer(&scorer) {
-							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
+							log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e);
+							// We opt not to abort early on persistence failure here as persisting
+							// the scorer is non-critical and we still hope that it will have
+							// resolved itself when it is potentially critical in event handling
+							// below.
 						}
 					}
 				}
 			}
-			event_handler(event).await;
+			event_handler(event).await
 		})
 	};
 	define_run_body!(
@@ -913,7 +920,7 @@ impl BackgroundProcessor {
 						}
 					}
 				}
-				event_handler.handle_event(event);
+				event_handler.handle_event(event)
 			};
 			define_run_body!(
 				persister,
@@ -1012,10 +1019,13 @@ mod tests {
 	use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
 	use bitcoin::transaction::Version;
 	use bitcoin::{Amount, ScriptBuf, Txid};
+	use core::sync::atomic::{AtomicBool, Ordering};
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::chain::transaction::OutPoint;
 	use lightning::chain::{chainmonitor, BestBlock, Confirm, Filter};
-	use lightning::events::{Event, MessageSendEvent, MessageSendEventsProvider, PathFailure};
+	use lightning::events::{
+		Event, MessageSendEvent, MessageSendEventsProvider, PathFailure, ReplayEvent,
+	};
 	use lightning::ln::channelmanager;
 	use lightning::ln::channelmanager::{
 		ChainParameters, PaymentId, BREAKDOWN_TIMEOUT, MIN_CLTV_EXPIRY_DELTA,
@@ -1757,7 +1767,7 @@ mod tests {
 		// Initiate the background processors to watch each node.
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister = Arc::new(Persister::new(data_dir));
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -1847,7 +1857,7 @@ mod tests {
 		let (_, nodes) = create_nodes(1, "test_timer_tick_called");
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister = Arc::new(Persister::new(data_dir));
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -1889,7 +1899,7 @@ mod tests {
 		let persister = Arc::new(
 			Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"),
 		);
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -1924,7 +1934,7 @@ mod tests {
 
 		let bp_future = super::process_events_async(
 			persister,
-			|_: _| async {},
+			|_: _| async { Ok(()) },
 			nodes[0].chain_monitor.clone(),
 			nodes[0].node.clone(),
 			Some(nodes[0].messenger.clone()),
@@ -1957,7 +1967,7 @@ mod tests {
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister =
 			Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test"));
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -1986,7 +1996,7 @@ mod tests {
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister =
 			Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test"));
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -2021,13 +2031,16 @@ mod tests {
 		// Set up a background event handler for FundingGenerationReady events.
 		let (funding_generation_send, funding_generation_recv) = std::sync::mpsc::sync_channel(1);
 		let (channel_pending_send, channel_pending_recv) = std::sync::mpsc::sync_channel(1);
-		let event_handler = move |event: Event| match event {
-			Event::FundingGenerationReady { .. } => funding_generation_send
-				.send(handle_funding_generation_ready!(event, channel_value))
-				.unwrap(),
-			Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
-			Event::ChannelReady { .. } => {},
-			_ => panic!("Unexpected event: {:?}", event),
+		let event_handler = move |event: Event| {
+			match event {
+				Event::FundingGenerationReady { .. } => funding_generation_send
+					.send(handle_funding_generation_ready!(event, channel_value))
+					.unwrap(),
+				Event::ChannelPending { .. } => channel_pending_send.send(()).unwrap(),
+				Event::ChannelReady { .. } => {},
+				_ => panic!("Unexpected event: {:?}", event),
+			}
+			Ok(())
 		};
 
 		let bg_processor = BackgroundProcessor::start(
@@ -2082,11 +2095,14 @@ mod tests {
 
 		// Set up a background event handler for SpendableOutputs events.
 		let (sender, receiver) = std::sync::mpsc::sync_channel(1);
-		let event_handler = move |event: Event| match event {
-			Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
-			Event::ChannelReady { .. } => {},
-			Event::ChannelClosed { .. } => {},
-			_ => panic!("Unexpected event: {:?}", event),
+		let event_handler = move |event: Event| {
+			match event {
+				Event::SpendableOutputs { .. } => sender.send(event).unwrap(),
+				Event::ChannelReady { .. } => {},
+				Event::ChannelClosed { .. } => {},
+				_ => panic!("Unexpected event: {:?}", event),
+			}
+			Ok(())
 		};
 		let persister = Arc::new(Persister::new(data_dir));
 		let bg_processor = BackgroundProcessor::start(
@@ -2215,12 +2231,60 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn test_event_handling_failures_are_replayed() {
+		let (_, nodes) = create_nodes(2, "test_event_handling_failures_are_replayed");
+		let channel_value = 100000;
+		let data_dir = nodes[0].kv_store.get_data_dir();
+		let persister = Arc::new(Persister::new(data_dir.clone()));
+
+		let (first_event_send, first_event_recv) = std::sync::mpsc::sync_channel(1);
+		let (second_event_send, second_event_recv) = std::sync::mpsc::sync_channel(1);
+		let should_fail_event_handling = Arc::new(AtomicBool::new(true));
+		let event_handler = move |event: Event| {
+			if let Ok(true) = should_fail_event_handling.compare_exchange(
+				true,
+				false,
+				Ordering::Acquire,
+				Ordering::Relaxed,
+			) {
+				first_event_send.send(event).unwrap();
+				return Err(ReplayEvent());
+			}
+
+			second_event_send.send(event).unwrap();
+			Ok(())
+		};
+
+		let bg_processor = BackgroundProcessor::start(
+			persister,
+			event_handler,
+			nodes[0].chain_monitor.clone(),
+			nodes[0].node.clone(),
+			Some(nodes[0].messenger.clone()),
+			nodes[0].no_gossip_sync(),
+			nodes[0].peer_manager.clone(),
+			nodes[0].logger.clone(),
+			Some(nodes[0].scorer.clone()),
+		);
+
+		begin_open_channel!(nodes[0], nodes[1], channel_value);
+		assert_eq!(
+			first_event_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE)),
+			second_event_recv.recv_timeout(Duration::from_secs(EVENT_DEADLINE))
+		);
+
+		if !std::thread::panicking() {
+			bg_processor.stop().unwrap();
+		}
+	}
+
 	#[test]
 	fn test_scorer_persistence() {
 		let (_, nodes) = create_nodes(2, "test_scorer_persistence");
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister = Arc::new(Persister::new(data_dir));
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let bg_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -2315,7 +2379,7 @@ mod tests {
 		let data_dir = nodes[0].kv_store.get_data_dir();
 		let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender));
 
-		let event_handler = |_: _| {};
+		let event_handler = |_: _| Ok(());
 		let background_processor = BackgroundProcessor::start(
 			persister,
 			event_handler,
@@ -2350,7 +2414,7 @@ mod tests {
 		let (exit_sender, exit_receiver) = tokio::sync::watch::channel(());
 		let bp_future = super::process_events_async(
 			persister,
-			|_: _| async {},
+			|_: _| async { Ok(()) },
 			nodes[0].chain_monitor.clone(),
 			nodes[0].node.clone(),
 			Some(nodes[0].messenger.clone()),
@@ -2492,12 +2556,15 @@ mod tests {
 	#[test]
 	fn test_payment_path_scoring() {
 		let (sender, receiver) = std::sync::mpsc::sync_channel(1);
-		let event_handler = move |event: Event| match event {
-			Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
-			Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
-			Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
-			Event::ProbeFailed { .. } => sender.send(event).unwrap(),
-			_ => panic!("Unexpected event: {:?}", event),
+		let event_handler = move |event: Event| {
+			match event {
+				Event::PaymentPathFailed { .. } => sender.send(event).unwrap(),
+				Event::PaymentPathSuccessful { .. } => sender.send(event).unwrap(),
+				Event::ProbeSuccessful { .. } => sender.send(event).unwrap(),
+				Event::ProbeFailed { .. } => sender.send(event).unwrap(),
+				_ => panic!("Unexpected event: {:?}", event),
+			}
+			Ok(())
 		};
 
 		let (_, nodes) = create_nodes(1, "test_payment_path_scoring");
@@ -2543,6 +2610,7 @@ mod tests {
 					Event::ProbeFailed { .. } => sender_ref.send(event).await.unwrap(),
 					_ => panic!("Unexpected event: {:?}", event),
 				}
+				Ok(())
 			}
 		};
 

lightning-invoice/src/utils.rs

@@ -1391,6 +1391,7 @@ mod test {
 			} else {
 				other_events.borrow_mut().push(event);
 			}
+			Ok(())
 		};
 		nodes[fwd_idx].node.process_pending_events(&forward_event_handler);
 		nodes[fwd_idx].node.process_pending_events(&forward_event_handler);

lightning/src/chain/chainmonitor.rs

@@ -33,8 +33,7 @@ use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::ln::types::ChannelId;
 use crate::sign::ecdsa::EcdsaChannelSigner;
-use crate::events;
-use crate::events::{Event, EventHandler};
+use crate::events::{self, Event, EventHandler, ReplayEvent};
 use crate::util::logger::{Logger, WithContext};
 use crate::util::errors::APIError;
 use crate::util::wakers::{Future, Notifier};
@@ -533,7 +532,7 @@ where C::Target: chain::Filter,
 	pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
 		use crate::events::EventsProvider;
 		let events = core::cell::RefCell::new(Vec::new());
-		let event_handler = |event: events::Event| events.borrow_mut().push(event);
+		let event_handler = |event: events::Event| Ok(events.borrow_mut().push(event));
 		self.process_pending_events(&event_handler);
 		events.into_inner()
 	}
@@ -544,16 +543,21 @@ where C::Target: chain::Filter,
 	/// See the trait-level documentation of [`EventsProvider`] for requirements.
 	///
 	/// [`EventsProvider`]: crate::events::EventsProvider
-	pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(
+	pub async fn process_pending_events_async<Future: core::future::Future<Output = Result<(), ReplayEvent>>, H: Fn(Event) -> Future>(
 		&self, handler: H
 	) {
 		// Sadly we can't hold the monitors read lock through an async call. Thus we have to do a
 		// crazy dance to process a monitor's events then only remove them once we've done so.
 		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();
+				}
+			}
 		}
 	}
 
@@ -880,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();
+				}
+			}
 		}
 	}
 }