Add simple test for event replaying

lightning-background-processor/src/lib.rs

@@ -932,7 +932,7 @@ mod tests {
 	use lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 	use lightning::sign::{InMemorySigner, KeysManager, ChangeDestinationSource};
 	use lightning::chain::transaction::OutPoint;
-	use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent};
+	use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent, ReplayEvent};
 	use lightning::{get_event_msg, get_event};
 	use lightning::ln::types::{PaymentHash, ChannelId};
 	use lightning::ln::channelmanager;
@@ -954,6 +954,7 @@ mod tests {
 		SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY};
 	use lightning::util::sweep::{OutputSweeper, OutputSpendStatus};
 	use lightning_persister::fs_store::FilesystemStore;
+	use core::sync::atomic::{AtomicBool, Ordering};
 	use std::collections::VecDeque;
 	use std::{fs, env};
 	use std::path::PathBuf;
@@ -1774,6 +1775,40 @@ 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");