Process ChannelManager events in the background

Author: jkczyz

lightning-background-processor/src/lib.rs

@@ -14,6 +14,7 @@ use lightning::chain::keysinterface::{Sign, KeysInterface};
 use lightning::ln::channelmanager::ChannelManager;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
 use lightning::ln::peer_handler::{PeerManager, SocketDescriptor};
+use lightning::util::events::{EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -109,11 +110,12 @@ impl BackgroundProcessor {
 		Descriptor: 'static + SocketDescriptor + Send + Sync,
 		CMH: 'static + Deref + Send + Sync,
 		RMH: 'static + Deref + Send + Sync,
+		EH: 'static + EventHandler + Send + Sync,
 		CMP: 'static + Send + ChannelManagerPersister<Signer, M, T, K, F, L>,
 		CM: 'static + Deref<Target = ChannelManager<Signer, M, T, K, F, L>> + Send + Sync,
 		PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, L>> + Send + Sync,
 	>
-	(handler: CMP, channel_manager: CM, peer_manager: PM, logger: L) -> Self
+	(persister: CMP, event_handler: EH, channel_manager: CM, peer_manager: PM, logger: L) -> Self
 	where
 		M::Target: 'static + chain::Watch<Signer>,
 		T::Target: 'static + BroadcasterInterface,
@@ -129,10 +131,11 @@ impl BackgroundProcessor {
 			let mut current_time = Instant::now();
 			loop {
 				peer_manager.process_events();
+				channel_manager.process_pending_events(&event_handler);
 				let updates_available =
 					channel_manager.await_persistable_update_timeout(Duration::from_millis(100));
 				if updates_available {
-					handler.persist_manager(&*channel_manager)?;
+					persister.persist_manager(&*channel_manager)?;
 				}
 				// Exit the loop if the background processor was requested to stop.
 				if stop_thread.load(Ordering::Acquire) == true {
@@ -162,10 +165,8 @@ mod tests {
 	use bitcoin::blockdata::constants::genesis_block;
 	use bitcoin::blockdata::transaction::{Transaction, TxOut};
 	use bitcoin::network::constants::Network;
-	use lightning::chain;
-	use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 	use lightning::chain::chainmonitor;
-	use lightning::chain::keysinterface::{Sign, InMemorySigner, KeysInterface, KeysManager};
+	use lightning::chain::keysinterface::{InMemorySigner, KeysInterface, KeysManager};
 	use lightning::chain::transaction::OutPoint;
 	use lightning::get_event_msg;
 	use lightning::ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, SimpleArcChannelManager};
@@ -174,7 +175,6 @@ mod tests {
 	use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
 	use lightning::util::config::UserConfig;
 	use lightning::util::events::{Event, MessageSendEventsProvider, MessageSendEvent};
-	use lightning::util::logger::Logger;
 	use lightning::util::ser::Writeable;
 	use lightning::util::test_utils;
 	use lightning_persister::FilesystemPersister;
@@ -279,14 +279,15 @@ mod tests {
 		// re-persistence and is successfully re-persisted.
 		let nodes = create_nodes(2, "test_background_processor".to_string());
 
-		// Initiate the background processors to watch each node.
-		let data_dir = nodes[0].persister.get_data_dir();
-		let callback = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
-		let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
-
 		// Go through the channel creation process until each node should have something persisted.
 		let tx = open_channel!(nodes[0], nodes[1], 100000);
 
+		// Initiate the background processors to watch each node.
+		let data_dir = nodes[0].persister.get_data_dir();
+		let persister = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
+		let event_handler = |_| {};
+		let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
+
 		macro_rules! check_persisted_data {
 			($node: expr, $filepath: expr, $expected_bytes: expr) => {
 				match $node.write(&mut $expected_bytes) {
@@ -336,8 +337,9 @@ mod tests {
 		// `FRESHNESS_TIMER`.
 		let nodes = create_nodes(1, "test_timer_tick_called".to_string());
 		let data_dir = nodes[0].persister.get_data_dir();
-		let callback = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
-		let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
+		let persister = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
+		let event_handler = |_| {};
+		let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
 		loop {
 			let log_entries = nodes[0].logger.lines.lock().unwrap();
 			let desired_log = "Calling ChannelManager's and PeerManager's timer_tick_occurred".to_string();
@@ -352,21 +354,12 @@ mod tests {
 	#[test]
 	fn test_persist_error() {
 		// Test that if we encounter an error during manager persistence, the thread panics.
-		fn persist_manager<Signer, M, T, K, F, L>(_data: &ChannelManager<Signer, Arc<M>, Arc<T>, Arc<K>, Arc<F>, Arc<L>>) -> Result<(), std::io::Error>
-		where Signer: 'static + Sign,
-		      M: 'static + chain::Watch<Signer>,
-		      T: 'static + BroadcasterInterface,
-		      K: 'static + KeysInterface<Signer=Signer>,
-		      F: 'static + FeeEstimator,
-		      L: 'static + Logger,
-		{
-			Err(std::io::Error::new(std::io::ErrorKind::Other, "test"))
-		}
-
 		let nodes = create_nodes(2, "test_persist_error".to_string());
-		let bg_processor = BackgroundProcessor::start(persist_manager, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
 		open_channel!(nodes[0], nodes[1], 100000);
 
+		let persister = |_: &_| Err(std::io::Error::new(std::io::ErrorKind::Other, "test"));
+		let event_handler = |_| {};
+		let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
 		let _ = bg_processor.thread_handle.join().unwrap().expect_err("Errored persisting manager: test");
 	}
 }