Shutdown: Wait for event processing to fully stop

.. before initiating the Runtime shutdown.

Author: tnull

src/builder.rs

@@ -959,6 +959,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));
@@ -971,6 +972,7 @@ fn build_with_store_internal(
 	Ok(Node {
 		runtime,
 		stop_sender,
+		event_handling_stopped_sender,
 		config,
 		wallet,
 		tx_sync,

src/lib.rs

@@ -173,6 +173,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>>>,
@@ -731,6 +732,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,
@@ -751,6 +753,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() {
@@ -800,9 +814,55 @@ 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();
+
+		// FIXME: For now, we wait up to 100 secs (BDK_WALLET_SYNC_TIMEOUT_SECS + 10) to allow
+		// event handling to exit gracefully even if it was blocked on the BDK wallet syncing. We
+		// should drop this considerably post upgrading to BDK 1.0.
+		let timeout_res = runtime.block_on(async {
+			tokio::time::timeout(
+				Duration::from_secs(100),
+				event_handling_stopped_receiver.changed(),
+			)
+			.await
+		});
+
+		match timeout_res {
+			Ok(stop_res) => match stop_res {
+				Ok(()) => {},
+				Err(e) => {
+					log_error!(
+						event_handling_stopped_logger,
+						"Stopping event handling failed. This should never happen: {}",
+						e
+					);
+					panic!("Stopping event handling failed. This should never happen.");
+				},
+			},
+			Err(e) => {
+				log_error!(
+					event_handling_stopped_logger,
+					"Stopping event handling timed out: {}",
+					e
+				);
+			},
+		}
+
+		#[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.");