Wind down spawned tasks when startup fails

src/lib.rs

@@ -302,7 +302,10 @@ impl Node {
 		match self.start_inner(&mut is_running_lock) {
 			Ok(()) => Ok(()),
 			Err(e) => {
-				self.chain_source.stop();
+				// Startup spawns background tasks before it can fail, e.g., when binding our
+				// listening addresses. Wind them all back down rather than leaving them running
+				// behind a node that never came up.
+				self.shutdown();
 				Err(e)
 			},
 		}
@@ -852,24 +855,29 @@ impl Node {
 
 		log_info!(self.logger, "Shutting down LDK Node with node ID {}...", self.node_id());
 
+		self.shutdown();
+
+		log_info!(self.logger, "Shutdown complete.");
+		*is_running_lock = false;
+		Ok(())
+	}
+
+	/// Winds down everything [`Node::start_inner`] may have brought up.
+	///
+	/// Unlike [`Node::stop`], this makes no assumption about how far startup progressed: it is
+	/// also used to clean up after a [`Node::start`] that failed part-way through, in which case
+	/// some of the background tasks below were never spawned and the shutdown signals accordingly
+	/// find no receivers.
+	fn shutdown(&self) {
 		// Prevent blocking Electrum syncs from making any further callbacks before persistence
 		// tasks stop accepting work.
 		self.chain_source.begin_shutdown();
 
 		// Stop background tasks.
-		self.stop_sender
-			.send(())
-			.map(|_| {
-				log_trace!(self.logger, "Sent shutdown signal to background tasks.");
-			})
-			.unwrap_or_else(|e| {
-				log_error!(
-					self.logger,
-					"Failed to send shutdown signal. This should never happen: {}",
-					e
-				);
-				debug_assert!(false);
-			});
+		match self.stop_sender.send(()) {
+			Ok(()) => log_trace!(self.logger, "Sent shutdown signal to background tasks."),
+			Err(_) => log_trace!(self.logger, "No background tasks to signal shutdown to."),
+		}
 
 		// Cancel cancellable background tasks
 		self.runtime.abort_cancellable_background_tasks();
@@ -886,29 +894,16 @@ impl Node {
 		log_debug!(self.logger, "Stopped chain sources.");
 
 		// Stop the background processor.
-		self.background_processor_stop_sender
-			.send(())
-			.map(|_| {
-				log_trace!(self.logger, "Sent shutdown signal to background processor.");
-			})
-			.unwrap_or_else(|e| {
-				log_error!(
-					self.logger,
-					"Failed to send shutdown signal. This should never happen: {}",
-					e
-				);
-				debug_assert!(false);
-			});
+		match self.background_processor_stop_sender.send(()) {
+			Ok(()) => log_trace!(self.logger, "Sent shutdown signal to background processor."),
+			Err(_) => log_trace!(self.logger, "No background processor to signal shutdown to."),
+		}
 
 		// Finally, wait until background processing stopped, at least until a timeout is reached.
 		self.runtime.wait_on_background_processor_task();
 
 		#[cfg(tokio_unstable)]
 		self.runtime.log_metrics();
-
-		log_info!(self.logger, "Shutdown complete.");
-		*is_running_lock = false;
-		Ok(())
 	}
 
 	/// Returns the status of the [`Node`].

src/runtime.rs

@@ -199,9 +199,12 @@ impl Runtime {
 		self.block_on(tasks.wait())
 	}
 
+	/// Waits for all non-cancellable background tasks to finish.
+	///
+	/// Note this may find no tasks at all, as it's also reached when winding down a startup that
+	/// failed before spawning any.
 	pub fn wait_on_background_tasks(&self) {
 		let mut tasks = core::mem::take(&mut *self.background_tasks.lock().expect("lock"));
-		debug_assert!(tasks.len() > 0, "Expected some background_tasks");
 		self.block_on(async {
 			loop {
 				let timeout_fut = tokio::time::timeout(
@@ -231,6 +234,10 @@ impl Runtime {
 		})
 	}
 
+	/// Waits for the background processor task to finish.
+	///
+	/// Note this may find no task at all, as it's also reached when winding down a startup that
+	/// failed before spawning it.
 	pub fn wait_on_background_processor_task(&self) {
 		if let Some(background_processor_task) =
 			self.background_processor_task.lock().expect("lock").take()
@@ -265,9 +272,7 @@ impl Runtime {
 					log_error!(self.logger, "Stopping event handling timed out: {}", e);
 				},
 			}
-		} else {
-			debug_assert!(false, "Expected a background processing task");
-		};
+		}
 	}
 
 	#[cfg(tokio_unstable)]
@@ -460,6 +465,16 @@ mod tests {
 		);
 	}
 
+	#[test]
+	fn winding_down_without_spawned_tasks_is_a_noop() {
+		// A `Node::start` that fails before spawning anything still runs the full shutdown
+		// sequence, so the wind-down has to tolerate finding nothing to wait on.
+		let runtime = test_runtime();
+		runtime.abort_cancellable_background_tasks();
+		runtime.wait_on_background_tasks();
+		runtime.wait_on_background_processor_task();
+	}
+
 	#[test]
 	fn late_cancellable_spawns_are_not_polled_after_abort() {
 		let runtime = test_runtime();

tests/integration_tests_rust.rs

@@ -918,6 +918,46 @@ async fn start_stop_with_pathfinding_scores_sync() {
 	node.stop().unwrap();
 }
 
+// A `start` that fails part-way through has to wind down whatever it already spawned. Here we take
+// one of the node's listening addresses before starting, so binding it fails only after the
+// wallet-sync and pathfinding-scores tasks are up.
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn failed_start_winds_down_background_tasks() {
+	let (_bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let config = random_config();
+
+	let listening_address =
+		config.node_config.listening_addresses.as_ref().unwrap().first().unwrap().to_string();
+	let squatter = std::net::TcpListener::bind(&listening_address).unwrap();
+
+	let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
+
+	let log_writer = Arc::new(CollectingLogWriter::new());
+	let mut sync_config = EsploraSyncConfig::default();
+	sync_config.background_sync_config = None;
+	setup_builder!(builder, config.node_config);
+	builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config));
+	builder.set_pathfinding_scores_source(esplora_url);
+	builder.set_custom_logger(log_writer.clone());
+
+	let node = builder.build(config.node_entropy.into()).unwrap();
+
+	assert_eq!(node.start(), Err(NodeError::InvalidSocketAddress));
+	assert!(!node.status().is_running);
+	assert_eq!(node.stop(), Err(NodeError::NotRunning));
+
+	// The failed startup ran the full shutdown sequence, rather than leaving the tasks it had
+	// already spawned running behind a node that never came up.
+	assert!(log_writer.contains("Stopped all background tasks"));
+	assert!(log_writer.contains("Disconnected all network peers."));
+	assert!(log_writer.contains("Stopped chain sources."));
+
+	// Having wound everything down, the node comes up cleanly once the address is free again.
+	drop(squatter);
+	node.start().unwrap();
+	node.stop().unwrap();
+}
+
 // The Electrum chain source drops its runtime client - and with it the tx-sync client holding all
 // `Filter` registrations - when stopped. As `ChannelMonitor`s only register their watched
 // transactions and outputs while being loaded in `Builder::build`, nothing would re-register them