Skip to content

Release completed cancellable tasks #997

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tnull wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Release completed cancellable tasks #997

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
08d3b5d
Release completed cancellable tasks
tnull Jul 20, 2026
8304c3f
Clean up abandoned liquidity requests
tnull Jul 20, 2026
5173b2b
Clean up cancelled connection attempts
tnull Jul 20, 2026
fb5fbc8
Reset cancelled wallet sync state
tnull Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ base64 = { version = "0.22.1", default-features = false, features = ["std"] }
getrandom = { version = "0.3", default-features = false }
chrono = { version = "0.4", default-features = false, features = ["clock"] }
tokio = { version = "1.39", default-features = false, features = [ "rt-multi-thread", "time", "sync", "macros", "net" ] }
tokio-util = { version = "0.7", default-features = false, features = ["rt"] }
esplora-client = { version = "0.12", default-features = false, features = ["tokio", "async-https-rustls"] }
electrum-client = { version = "0.25", default-features = false, features = ["proxy", "use-rustls-ring"] }
libc = "0.2"
Expand Down
103 changes: 91 additions & 12 deletions src/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ use std::time::Duration;

use lightning::util::native_async::FutureSpawner;
use tokio::task::{JoinHandle, JoinSet};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;

use crate::config::{
BACKGROUND_TASK_SHUTDOWN_TIMEOUT_SECS, LDK_EVENT_HANDLER_SHUTDOWN_TIMEOUT_SECS,
Expand All @@ -28,13 +30,18 @@ pub(crate) struct Runtime {
}

struct CancellableBackgroundTasks {
tasks: JoinSet<()>,
tasks: TaskTracker,
cancellation_token: CancellationToken,
accepting_tasks: bool,

@tankyleo tankyleo Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can delete this field here, and track is_closed on TaskTracker instead

}

impl CancellableBackgroundTasks {
fn new() -> Self {
Self { tasks: JoinSet::new(), accepting_tasks: true }
Self {
tasks: TaskTracker::new(),
cancellation_token: CancellationToken::new(),
accepting_tasks: true,
}
}
}

Expand Down Expand Up @@ -109,8 +116,7 @@ impl Runtime {
where
F: Future<Output = ()> + Send + 'static,
{
let mut cancellable_background_tasks =
self.cancellable_background_tasks.lock().expect("lock");
let cancellable_background_tasks = self.cancellable_background_tasks.lock().expect("lock");
if !cancellable_background_tasks.accepting_tasks {
log_trace!(
self.logger,
Expand All @@ -122,11 +128,32 @@ impl Runtime {
// Since it seems to make a difference to `tokio` (see
// https://docs.rs/tokio/latest/tokio/time/fn.timeout.html#panics) we make sure the futures
// are always put in an `async` / `.await` closure.
cancellable_background_tasks.tasks.spawn_on(async { future.await }, runtime_handle);
let cancellation_token = cancellable_background_tasks.cancellation_token.clone();
// Detach the handle while the tracker continues tracking the task.
let _ = cancellable_background_tasks.tasks.spawn_on(

@tankyleo tankyleo Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Codex:

  • [P2] Preserve abort-on-drop for detached cancellable tasks _ /home/ubuntu/ldk-node/src/runtime.rs:133-133
    Dropping the JoinHandle here changes the old JoinSet drop behavior: TaskTracker explicitly does not abort tracked tasks when it is dropped. With an externally supplied Tokio runtime, if Node::start() returns an error after spawning cancellable tasks but before is_running is set, Drop won't call
    stop(), and these detached tasks can keep running and holding node Arcs after the node is dropped instead of being aborted as before. Please retain abort-on-drop handles or cancel/abort tracked tasks from Runtime's drop path.

This might be related to the PR we've had in mind about proper clean up after fn start

async move {
tokio::select! {
biased;
_ = cancellation_token.cancelled() => {},
_ = future => {},
}
},
runtime_handle,
);
}

pub fn allow_cancellable_background_task_spawns(&self) {
self.cancellable_background_tasks.lock().expect("lock").accepting_tasks = true;
let mut cancellable_background_tasks =
self.cancellable_background_tasks.lock().expect("lock");
if cancellable_background_tasks.cancellation_token.is_cancelled() {
debug_assert!(

@tankyleo tankyleo Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe this debug_assert is reachable here: further below we cancel the token, then drop the lock, then wait. So here the token could be canceled, but the tasks not yet empty.

@tnull tnull Jul 27, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, it's def. not reachable from the current callsites as they are serialized.

@tankyleo tankyleo Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed the current callsites are serialized. Might be good to not rely on this external lock for serialization, I vibed with codex on this issue, here's the patch it came up with below. Your call if you want to incorporate this or leave the PR as is :)

    Isolate cancellable task generations

    Replace cancelled cancellable-task state as a unit so a restart cannot reopen the tracker being drained by an in-flight abort.

    AI-Assisted-By: OpenAI Codex

diff --git a/src/runtime.rs b/src/runtime.rs
index d4ea7ce..3fe04d1 100644
--- a/src/runtime.rs
+++ b/src/runtime.rs
@@ -146,14 +146,10 @@ impl Runtime {
                let mut cancellable_background_tasks =
                        self.cancellable_background_tasks.lock().expect("lock");
                if cancellable_background_tasks.cancellation_token.is_cancelled() {
-                       debug_assert!(
-                               cancellable_background_tasks.tasks.is_empty(),
-                               "Expected all cancellable background tasks to be stopped"
-                       );
-                       cancellable_background_tasks.cancellation_token = CancellationToken::new();
+                       // An abort may still be waiting on a clone of the previous task tracker. Start a new
+                       // generation instead of reopening that tracker underneath the in-flight wait.
+                       *cancellable_background_tasks = CancellableBackgroundTasks::new();
                }
-               cancellable_background_tasks.tasks.reopen();
-               cancellable_background_tasks.accepting_tasks = true;
        }

        pub fn spawn_background_processor_task<F>(&self, future: F)

cancellable_background_tasks.tasks.is_empty(),
"Expected all cancellable background tasks to be stopped"
);
cancellable_background_tasks.cancellation_token = CancellationToken::new();
}
cancellable_background_tasks.tasks.reopen();
cancellable_background_tasks.accepting_tasks = true;
}

pub fn spawn_background_processor_task<F>(&self, future: F)
Expand Down Expand Up @@ -164,15 +191,15 @@ impl Runtime {
}

pub fn abort_cancellable_background_tasks(&self) {
let mut tasks = {
let tasks = {
let mut cancellable_background_tasks =
self.cancellable_background_tasks.lock().expect("lock");
cancellable_background_tasks.accepting_tasks = false;
core::mem::take(&mut cancellable_background_tasks.tasks)
cancellable_background_tasks.tasks.close();
cancellable_background_tasks.cancellation_token.cancel();
cancellable_background_tasks.tasks.clone()
};
debug_assert!(tasks.len() > 0, "Expected some cancellable background_tasks");
tasks.abort_all();
self.block_on(async { while let Some(_) = tasks.join_next().await {} })
self.block_on(tasks.wait())

@tankyleo tankyleo Jul 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIRC while we are waiting on tasks.wait, we could call allow_cancellable_background_task_spawns, reopen the TaskTracker, and thus tasks.wait() would never finish.

@tnull tnull Jul 27, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, yes, if we only look at Runtime alone, but note that we are always acquiring the is_running lock in start/stop. Let me know if you think it's crucial to make Runtime more robust by itself.

@tankyleo tankyleo Jul 27, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let me know what you think of the patch above, it would address this here too.

}

pub fn wait_on_background_tasks(&self) {
Expand Down Expand Up @@ -381,19 +408,68 @@ impl FutureSpawner for RuntimeSpawner {

#[cfg(test)]
mod tests {
use tokio::sync::oneshot;
use tokio::sync::{mpsc, oneshot};

use super::*;

struct DropNotifier(Option<oneshot::Sender<()>>);

impl Drop for DropNotifier {
fn drop(&mut self) {
if let Some(sender) = self.0.take() {
let _ = sender.send(());
}
}
}

fn test_runtime() -> Runtime {
Runtime::new(Arc::new(Logger::new_log_facade())).unwrap()
}

#[test]
fn completed_cancellable_tasks_are_released_before_shutdown() {
const TASK_COUNT: usize = 64;

let runtime = test_runtime();
let (completion_sender, mut completion_receiver) = mpsc::channel(TASK_COUNT);
for _ in 0..TASK_COUNT {
let completion_sender = completion_sender.clone();
runtime.spawn_cancellable_background_task(async move {
completion_sender.send(()).await.expect("completion receiver should be open");
});
}
drop(completion_sender);

let completed_tasks_are_released = runtime.block_on(async {
for _ in 0..TASK_COUNT {
completion_receiver.recv().await.expect("cancellable task should complete");
}

tokio::time::timeout(Duration::from_secs(1), async {
loop {
if runtime.cancellable_background_tasks.lock().expect("lock").tasks.is_empty() {
break;
}
tokio::task::yield_now().await;
}
})
.await
.is_ok()
});

assert!(
completed_tasks_are_released,
"completed cancellable tasks should be released before shutdown"
);
}

#[test]
fn late_cancellable_spawns_are_not_polled_after_abort() {
let runtime = test_runtime();
let (started_sender, started_receiver) = oneshot::channel();
let (dropped_sender, dropped_receiver) = oneshot::channel();
runtime.spawn_cancellable_background_task(async move {
let _drop_notifier = DropNotifier(Some(dropped_sender));
let _ = started_sender.send(());
std::future::pending::<()>().await;
});
Expand All @@ -402,6 +478,9 @@ mod tests {
});

runtime.abort_cancellable_background_tasks();
runtime.block_on(async {
dropped_receiver.await.expect("aborted task should be dropped before abort returns");
});

let (late_spawn_sender, late_spawn_receiver) = oneshot::channel();
runtime.spawn_cancellable_background_task(async move {
Expand Down