Shutdown: Wait for event processing to fully stop

.. before initiating the Runtime shutdown.

Author: tnull

src/builder.rs

@@ -972,6 +972,7 @@ fn build_with_store_internal(
 	};
 
 	let (stop_sender, _) = tokio::sync::watch::channel(());
+	let (event_handling_stopped_sender, _) = tokio::sync::watch::channel(());
 
 	let is_listening = Arc::new(AtomicBool::new(false));
 	let latest_wallet_sync_timestamp = Arc::new(RwLock::new(None));
@@ -983,6 +984,7 @@ fn build_with_store_internal(
 	Ok(Node {
 		runtime,
 		stop_sender,
+		event_handling_stopped_sender,
 		config,
 		wallet,
 		tx_sync,

src/lib.rs

@@ -171,6 +171,7 @@ uniffi::include_scaffolding!("ldk_node");
 pub struct Node {
 	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	stop_sender: tokio::sync::watch::Sender<()>,
+	event_handling_stopped_sender: tokio::sync::watch::Sender<()>,
 	config: Arc<Config>,
 	wallet: Arc<Wallet>,
 	tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
@@ -714,6 +715,7 @@ impl Node {
 		};
 
 		let background_stop_logger = Arc::clone(&self.logger);
+		let event_handling_stopped_sender = self.event_handling_stopped_sender.clone();
 		runtime.spawn(async move {
 			process_events_async(
 				background_persister,
@@ -734,6 +736,18 @@ impl Node {
 				panic!("Failed to process events");
 			});
 			log_trace!(background_stop_logger, "Events processing stopped.",);
+
+			match event_handling_stopped_sender.send(()) {
+				Ok(_) => (),
+				Err(e) => {
+					log_error!(
+						background_stop_logger,
+						"Failed to send 'events handling stopped' signal. This should never happen: {}",
+						e
+						);
+					debug_assert!(false);
+				},
+			}
 		});
 
 		if let Some(liquidity_source) = self.liquidity_source.as_ref() {
@@ -783,9 +797,47 @@ impl Node {
 			},
 		}
 
-		// Stop disconnect peers.
+		// Disconnect all peers.
 		self.peer_manager.disconnect_all_peers();
 
+		// Wait until event handling stopped, at least until a timeout is reached.
+		let event_handling_stopped_logger = Arc::clone(&self.logger);
+		let mut event_handling_stopped_receiver = self.event_handling_stopped_sender.subscribe();
+
+		let _ = runtime
+			.block_on(async {
+				tokio::time::timeout(
+					Duration::from_secs(10),
+					event_handling_stopped_receiver.changed(),
+				)
+				.await
+			})
+			.map_err(|e| {
+				log_error!(
+					event_handling_stopped_logger,
+					"Stopping event handling timed out. This should never happen: {}",
+					e
+				);
+				debug_assert!(false);
+			})
+			.unwrap_or_else(|_| {
+				log_error!(
+					event_handling_stopped_logger,
+					"Stopping event handling failed. This should never happen.",
+				);
+				panic!("Stopping event handling failed. This should never happen.");
+			});
+
+		#[cfg(tokio_unstable)]
+		{
+			log_trace!(
+				self.logger,
+				"Active runtime tasks left prior to shutdown: {}",
+				runtime.metrics().active_tasks_count()
+			);
+		}
+
+		// Shutdown our runtime. By now ~no or only very few tasks should be left.
 		runtime.shutdown_timeout(Duration::from_secs(10));
 
 		log_info!(self.logger, "Shutdown complete.");