Skip to content

Commit 6d702c3

Browse files
Jolah1claude
andcommitted
Wind down spawned tasks when startup fails
`Node::start` spawns background tasks - wallet sync, RGS gossip, pathfinding scores - before it can still fail, e.g. when resolving or binding the configured listening addresses. Until now the error path only stopped the chain source, leaving those tasks running behind a node that never came up, and leaving the node in a state a subsequent `start` could not cleanly recover from. Extract the wind-down sequence from `Node::stop` into a `Node::shutdown` helper and run it on any `start_inner` error. As the helper now also runs after a partial startup, it can no longer assume that every task exists: the two shutdown `watch::Sender::send` calls are allowed to find no receivers, and the `debug_assert!`s in `Runtime::wait_on_background_tasks` and `Runtime::wait_on_background_processor_task` that required a fully-started node are dropped in favour of doc comments spelling out that case. Fixes #1009. This change was written with the assistance of Claude Code. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SUGgjhnpkCuFjsE3BjYAEx Claude-Session: https://claude.ai/code/session_01JRikhHBQigjpBY1FAc1yBr
1 parent b1337d2 commit 6d702c3

3 files changed

Lines changed: 90 additions & 35 deletions

File tree

src/lib.rs

Lines changed: 26 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -302,7 +302,10 @@ impl Node {
302302
match self.start_inner(&mut is_running_lock) {
303303
Ok(()) => Ok(()),
304304
Err(e) => {
305-
self.chain_source.stop();
305+
// Startup spawns background tasks before it can fail, e.g., when binding our
306+
// listening addresses. Wind them all back down rather than leaving them running
307+
// behind a node that never came up.
308+
self.shutdown();
306309
Err(e)
307310
},
308311
}
@@ -852,24 +855,29 @@ impl Node {
852855

853856
log_info!(self.logger, "Shutting down LDK Node with node ID {}...", self.node_id());
854857

858+
self.shutdown();
859+
860+
log_info!(self.logger, "Shutdown complete.");
861+
*is_running_lock = false;
862+
Ok(())
863+
}
864+
865+
/// Winds down everything [`Node::start_inner`] may have brought up.
866+
///
867+
/// Unlike [`Node::stop`], this makes no assumption about how far startup progressed: it is
868+
/// also used to clean up after a [`Node::start`] that failed part-way through, in which case
869+
/// some of the background tasks below were never spawned and the shutdown signals accordingly
870+
/// find no receivers.
871+
fn shutdown(&self) {
855872
// Prevent blocking Electrum syncs from making any further callbacks before persistence
856873
// tasks stop accepting work.
857874
self.chain_source.begin_shutdown();
858875

859876
// Stop background tasks.
860-
self.stop_sender
861-
.send(())
862-
.map(|_| {
863-
log_trace!(self.logger, "Sent shutdown signal to background tasks.");
864-
})
865-
.unwrap_or_else(|e| {
866-
log_error!(
867-
self.logger,
868-
"Failed to send shutdown signal. This should never happen: {}",
869-
e
870-
);
871-
debug_assert!(false);
872-
});
877+
match self.stop_sender.send(()) {
878+
Ok(()) => log_trace!(self.logger, "Sent shutdown signal to background tasks."),
879+
Err(_) => log_trace!(self.logger, "No background tasks to signal shutdown to."),
880+
}
873881

874882
// Cancel cancellable background tasks
875883
self.runtime.abort_cancellable_background_tasks();
@@ -886,29 +894,16 @@ impl Node {
886894
log_debug!(self.logger, "Stopped chain sources.");
887895

888896
// Stop the background processor.
889-
self.background_processor_stop_sender
890-
.send(())
891-
.map(|_| {
892-
log_trace!(self.logger, "Sent shutdown signal to background processor.");
893-
})
894-
.unwrap_or_else(|e| {
895-
log_error!(
896-
self.logger,
897-
"Failed to send shutdown signal. This should never happen: {}",
898-
e
899-
);
900-
debug_assert!(false);
901-
});
897+
match self.background_processor_stop_sender.send(()) {
898+
Ok(()) => log_trace!(self.logger, "Sent shutdown signal to background processor."),
899+
Err(_) => log_trace!(self.logger, "No background processor to signal shutdown to."),
900+
}
902901

903902
// Finally, wait until background processing stopped, at least until a timeout is reached.
904903
self.runtime.wait_on_background_processor_task();
905904

906905
#[cfg(tokio_unstable)]
907906
self.runtime.log_metrics();
908-
909-
log_info!(self.logger, "Shutdown complete.");
910-
*is_running_lock = false;
911-
Ok(())
912907
}
913908

914909
/// Returns the status of the [`Node`].

src/runtime.rs

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -199,9 +199,12 @@ impl Runtime {
199199
self.block_on(tasks.wait())
200200
}
201201

202+
/// Waits for all non-cancellable background tasks to finish.
203+
///
204+
/// Note this may find no tasks at all, as it's also reached when winding down a startup that
205+
/// failed before spawning any.
202206
pub fn wait_on_background_tasks(&self) {
203207
let mut tasks = core::mem::take(&mut *self.background_tasks.lock().expect("lock"));
204-
debug_assert!(tasks.len() > 0, "Expected some background_tasks");
205208
self.block_on(async {
206209
loop {
207210
let timeout_fut = tokio::time::timeout(
@@ -231,6 +234,10 @@ impl Runtime {
231234
})
232235
}
233236

237+
/// Waits for the background processor task to finish.
238+
///
239+
/// Note this may find no task at all, as it's also reached when winding down a startup that
240+
/// failed before spawning it.
234241
pub fn wait_on_background_processor_task(&self) {
235242
if let Some(background_processor_task) =
236243
self.background_processor_task.lock().expect("lock").take()
@@ -265,9 +272,7 @@ impl Runtime {
265272
log_error!(self.logger, "Stopping event handling timed out: {}", e);
266273
},
267274
}
268-
} else {
269-
debug_assert!(false, "Expected a background processing task");
270-
};
275+
}
271276
}
272277

273278
#[cfg(tokio_unstable)]
@@ -460,6 +465,16 @@ mod tests {
460465
);
461466
}
462467

468+
#[test]
469+
fn winding_down_without_spawned_tasks_is_a_noop() {
470+
// A `Node::start` that fails before spawning anything still runs the full shutdown
471+
// sequence, so the wind-down has to tolerate finding nothing to wait on.
472+
let runtime = test_runtime();
473+
runtime.abort_cancellable_background_tasks();
474+
runtime.wait_on_background_tasks();
475+
runtime.wait_on_background_processor_task();
476+
}
477+
463478
#[test]
464479
fn late_cancellable_spawns_are_not_polled_after_abort() {
465480
let runtime = test_runtime();

tests/integration_tests_rust.rs

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -918,6 +918,51 @@ async fn start_stop_with_pathfinding_scores_sync() {
918918
node.stop().unwrap();
919919
}
920920

921+
// A `start` that fails part-way through has to wind down whatever it already spawned. Here we take
922+
// one of the node's listening addresses before starting, so binding it fails only after the
923+
// wallet-sync, RGS and pathfinding-scores tasks have been spawned.
924+
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
925+
async fn failed_start_winds_down_background_tasks() {
926+
let (_bitcoind, electrsd) = setup_bitcoind_and_electrsd();
927+
let config = random_config();
928+
929+
let listening_address =
930+
config.node_config.listening_addresses.as_ref().unwrap().first().unwrap().to_string();
931+
let squatter = std::net::TcpListener::bind(&listening_address).unwrap();
932+
933+
let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
934+
935+
let log_writer = Arc::new(CollectingLogWriter::new());
936+
setup_builder!(builder, config.node_config);
937+
// Background syncing stays enabled, so the wallet-sync task is still running when binding
938+
// fails.
939+
builder.set_chain_source_esplora(esplora_url.clone(), None);
940+
builder.set_pathfinding_scores_source(esplora_url);
941+
// Nothing listens on port 1: the RGS task only needs to be spawned, not to succeed.
942+
builder.set_gossip_source_rgs("http://127.0.0.1:1".to_string());
943+
builder.set_custom_logger(log_writer.clone());
944+
945+
let node = builder.build(config.node_entropy.into()).unwrap();
946+
947+
assert_eq!(node.start(), Err(NodeError::InvalidSocketAddress));
948+
assert!(!node.status().is_running);
949+
assert_eq!(node.stop(), Err(NodeError::NotRunning));
950+
951+
// The failed startup ran the full shutdown sequence, rather than leaving the tasks it had
952+
// already spawned running behind a node that never came up.
953+
assert!(log_writer.contains("Stopped all background tasks"));
954+
assert!(log_writer.contains("Disconnected all network peers."));
955+
assert!(log_writer.contains("Stopped chain sources."));
956+
// The wallet-sync task only exits on the shutdown signal, so this shows a task that was
957+
// still running got stopped.
958+
assert!(log_writer.contains("Stopping background syncing on-chain wallet."));
959+
960+
// Having wound everything down, the node comes up cleanly once the address is free again.
961+
drop(squatter);
962+
node.start().unwrap();
963+
node.stop().unwrap();
964+
}
965+
921966
// The Electrum chain source drops its runtime client - and with it the tx-sync client holding all
922967
// `Filter` registrations - when stopped. As `ChannelMonitor`s only register their watched
923968
// transactions and outputs while being loaded in `Builder::build`, nothing would re-register them

0 commit comments

Comments
 (0)