From 5e7e8b82be909d50f7de2f8f2282c807b3e3f832 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 20 Jul 2026 10:40:48 +0200 Subject: [PATCH 01/10] Release completed cancellable tasks Avoid retaining completed Tokio task allocations for the node lifetime while preserving shutdown cancellation and restart semantics. Co-Authored-By: HAL 9000 --- Cargo.toml | 1 + src/runtime.rs | 103 +++++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 92 insertions(+), 12 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 8c793fb785..612128ee3b 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] } ldk-esplora-client = { package = "esplora-client", version = "0.13", default-features = false, features = ["tokio", "async-https-rustls"] } electrum-client = { version = "0.25", default-features = false, features = ["proxy", "use-rustls-ring"] } diff --git a/src/runtime.rs b/src/runtime.rs index 7e29996e62..d4ea7cea27 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -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, @@ -28,13 +30,18 @@ pub(crate) struct Runtime { } struct CancellableBackgroundTasks { - tasks: JoinSet<()>, + tasks: TaskTracker, + cancellation_token: CancellationToken, accepting_tasks: bool, } impl CancellableBackgroundTasks { fn new() -> Self { - Self { tasks: JoinSet::new(), accepting_tasks: true } + Self { + tasks: TaskTracker::new(), + cancellation_token: CancellationToken::new(), + accepting_tasks: true, + } } } @@ -109,8 +116,7 @@ impl Runtime { where F: Future + 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, @@ -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( + 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!( + 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(&self, future: F) @@ -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()) } pub fn wait_on_background_tasks(&self) { @@ -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>); + + 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; }); @@ -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 { From 553eeac5d147205fb225b9cfd2d09628a620705c Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 27 Jul 2026 15:40:24 +0200 Subject: [PATCH 02/10] f Use task tracker closure state Check the tracker directly instead of duplicating spawn state. This keeps task admission aligned with shutdown tracking. Co-Authored-By: HAL 9000 --- src/runtime.rs | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/src/runtime.rs b/src/runtime.rs index d4ea7cea27..af36a4dfa3 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -32,16 +32,11 @@ pub(crate) struct Runtime { struct CancellableBackgroundTasks { tasks: TaskTracker, cancellation_token: CancellationToken, - accepting_tasks: bool, } impl CancellableBackgroundTasks { fn new() -> Self { - Self { - tasks: TaskTracker::new(), - cancellation_token: CancellationToken::new(), - accepting_tasks: true, - } + Self { tasks: TaskTracker::new(), cancellation_token: CancellationToken::new() } } } @@ -117,7 +112,7 @@ impl Runtime { F: Future + Send + 'static, { let cancellable_background_tasks = self.cancellable_background_tasks.lock().expect("lock"); - if !cancellable_background_tasks.accepting_tasks { + if cancellable_background_tasks.tasks.is_closed() { log_trace!( self.logger, "Ignoring cancellable background task spawned during shutdown." @@ -153,7 +148,6 @@ impl Runtime { cancellable_background_tasks.cancellation_token = CancellationToken::new(); } cancellable_background_tasks.tasks.reopen(); - cancellable_background_tasks.accepting_tasks = true; } pub fn spawn_background_processor_task(&self, future: F) @@ -192,9 +186,8 @@ impl Runtime { pub fn abort_cancellable_background_tasks(&self) { let tasks = { - let mut cancellable_background_tasks = + let cancellable_background_tasks = self.cancellable_background_tasks.lock().expect("lock"); - cancellable_background_tasks.accepting_tasks = false; cancellable_background_tasks.tasks.close(); cancellable_background_tasks.cancellation_token.cancel(); cancellable_background_tasks.tasks.clone() From 595ce12a9c98b3125a24347848e47992b9bb4c83 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Wed, 5 Aug 2026 14:01:00 +0200 Subject: [PATCH 03/10] f Isolate cancellable task generations Replace closed task state as a unit so a restart cannot reopen the tracker being drained by an in-flight shutdown. Co-Authored-By: HAL 9000 --- src/runtime.rs | 75 +++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 68 insertions(+), 7 deletions(-) diff --git a/src/runtime.rs b/src/runtime.rs index af36a4dfa3..5d1d7625ef 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -140,14 +140,11 @@ impl Runtime { pub fn allow_cancellable_background_task_spawns(&self) { 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(); + if cancellable_background_tasks.tasks.is_closed() { + // 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(); } pub fn spawn_background_processor_task(&self, future: F) @@ -502,4 +499,68 @@ mod tests { }); runtime.abort_cancellable_background_tasks(); } + + #[test] + fn restart_does_not_reopen_draining_task_tracker() { + let tokio_runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("test runtime"); + let runtime = Arc::new(Runtime::with_handle( + tokio_runtime.handle().clone(), + Arc::new(Logger::new_log_facade()), + )); + + let (tracked_started_sender, tracked_started_receiver) = std::sync::mpsc::sync_channel(1); + runtime.spawn_cancellable_background_task(async move { + tracked_started_sender.send(()).expect("test should still be running"); + std::future::pending::<()>().await; + }); + tracked_started_receiver + .recv_timeout(Duration::from_secs(1)) + .expect("tracked task should start"); + + // Keep the only worker busy so cancellation cannot finish before a restart begins. + let blocker_barrier = Arc::new(std::sync::Barrier::new(2)); + let task_barrier = Arc::clone(&blocker_barrier); + let (blocker_started_sender, blocker_started_receiver) = std::sync::mpsc::sync_channel(1); + tokio_runtime.handle().spawn(async move { + blocker_started_sender.send(()).expect("test should still be running"); + task_barrier.wait(); + }); + blocker_started_receiver + .recv_timeout(Duration::from_secs(1)) + .expect("worker blocker should start"); + + let abort_runtime = Arc::clone(&runtime); + let abort_thread = + std::thread::spawn(move || abort_runtime.abort_cancellable_background_tasks()); + let deadline = std::time::Instant::now() + Duration::from_secs(1); + let draining_tasks = loop { + let tasks = runtime.cancellable_background_tasks.lock().expect("lock").tasks.clone(); + if tasks.is_closed() { + break tasks; + } + assert!(std::time::Instant::now() < deadline, "abort should close the task tracker"); + std::thread::yield_now(); + }; + + let restart_result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + runtime.allow_cancellable_background_task_spawns(); + })); + let draining_tracker_stayed_closed = draining_tasks.is_closed(); + // Ensure a broken implementation that reopens the old generation can still shut down. + if !draining_tracker_stayed_closed { + draining_tasks.close(); + } + blocker_barrier.wait(); + abort_thread.join().expect("abort should finish"); + + assert!( + restart_result.is_ok(), + "restart should use a new task tracker while the previous tracker drains" + ); + assert!(draining_tracker_stayed_closed, "restart should not reopen the draining tracker"); + } } From 2709fea40a8bb59d5761dcb111e26feb16409b49 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 10 Aug 2026 12:05:27 +0200 Subject: [PATCH 04/10] f Cancel detached tasks on runtime drop Signal cancellation when cancellable task state is dropped so detached tasks cannot outlive a node that fails during startup. Co-Authored-By: HAL 9000 --- src/runtime.rs | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/runtime.rs b/src/runtime.rs index 5d1d7625ef..5bff16b992 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -40,6 +40,13 @@ impl CancellableBackgroundTasks { } } +impl Drop for CancellableBackgroundTasks { + fn drop(&mut self) { + self.tasks.close(); + self.cancellation_token.cancel(); + } +} + impl Runtime { pub fn new(logger: Arc) -> Result { let mode = match tokio::runtime::Handle::try_current() { @@ -563,4 +570,34 @@ mod tests { ); assert!(draining_tracker_stayed_closed, "restart should not reopen the draining tracker"); } + + #[test] + fn dropping_runtime_cancels_detached_tasks() { + let tokio_runtime = + tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("test runtime"); + let runtime = Runtime::with_handle( + tokio_runtime.handle().clone(), + Arc::new(Logger::new_log_facade()), + ); + 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; + }); + tokio_runtime.block_on(async { + started_receiver.await.expect("task should start"); + }); + + drop(runtime); + + let task_was_dropped = tokio_runtime.block_on(async { + matches!( + tokio::time::timeout(Duration::from_secs(1), dropped_receiver).await, + Ok(Ok(())) + ) + }); + assert!(task_was_dropped, "dropping the runtime should cancel detached tasks"); + } } From 48beb4f17e1b9b623347a2740ea90523dd48c948 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 20 Jul 2026 11:09:45 +0200 Subject: [PATCH 05/10] Clean up abandoned liquidity requests Remove pending LSPS request state when callers time out or are cancelled so unresponsive services cannot grow request maps. Co-Authored-By: HAL 9000 --- src/liquidity/client/lsps1.rs | 55 +++++++----- src/liquidity/client/lsps2.rs | 38 +++++--- src/liquidity/mod.rs | 162 +++++++++++++++++++++++++++++----- 3 files changed, 198 insertions(+), 57 deletions(-) diff --git a/src/liquidity/client/lsps1.rs b/src/liquidity/client/lsps1.rs index 6082414c12..ef6ceea638 100644 --- a/src/liquidity/client/lsps1.rs +++ b/src/liquidity/client/lsps1.rs @@ -21,8 +21,8 @@ use tokio::sync::oneshot; use crate::connection::ConnectionManager; use crate::liquidity::{ - select_lsps_for_protocol, LspConfig, LspNode, LIQUIDITY_REQUEST_TIMEOUT_SECS, - LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, + select_lsps_for_protocol, LspConfig, LspNode, PendingRequest, PendingRequestGuard, + LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, }; use crate::logger::{log_error, log_info, LdkLogger, Logger}; use crate::runtime::Runtime; @@ -35,11 +35,11 @@ where { pub(crate) lsp_nodes: Arc>>, pub(crate) pending_opening_params_requests: - Mutex>>, + Mutex>>, pub(crate) pending_create_order_requests: - Mutex>>, + Mutex>>, pub(crate) pending_check_order_status_requests: - Mutex>>, + Mutex>>, pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, pub(crate) liquidity_manager: Arc, pub(crate) logger: L, @@ -61,12 +61,17 @@ where })?; let (request_sender, request_receiver) = oneshot::channel(); - { + let _pending_request = { let mut pending_opening_params_requests_lock = self.pending_opening_params_requests.lock().expect("lock"); let request_id = client_handler.request_supported_options(lsps1_node.node_id); - pending_opening_params_requests_lock.insert(request_id, request_sender); - } + PendingRequestGuard::insert( + &self.pending_opening_params_requests, + &mut pending_opening_params_requests_lock, + request_id, + request_sender, + ) + }; tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), request_receiver) .await @@ -146,7 +151,7 @@ where let (request_sender, request_receiver) = oneshot::channel(); let request_id; - { + let _pending_request = { let mut pending_create_order_requests_lock = self.pending_create_order_requests.lock().expect("lock"); request_id = client_handler.create_order( @@ -154,8 +159,13 @@ where order_params.clone(), Some(refund_address), ); - pending_create_order_requests_lock.insert(request_id.clone(), request_sender); - } + PendingRequestGuard::insert( + &self.pending_create_order_requests, + &mut pending_create_order_requests_lock, + request_id.clone(), + request_sender, + ) + }; let response = tokio::time::timeout( Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), @@ -191,12 +201,17 @@ where })?; let (request_sender, request_receiver) = oneshot::channel(); - { + let _pending_request = { let mut pending_check_order_status_requests_lock = self.pending_check_order_status_requests.lock().expect("lock"); let request_id = client_handler.check_order_status(&lsp_node_id, order_id); - pending_check_order_status_requests_lock.insert(request_id, request_sender); - } + PendingRequestGuard::insert( + &self.pending_check_order_status_requests, + &mut pending_check_order_status_requests_lock, + request_id, + request_sender, + ) + }; let response = tokio::time::timeout( Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), @@ -229,7 +244,7 @@ where .iter() .any(|n| n.node_id == counterparty_node_id) { - if let Some(sender) = self + if let Some(request) = self .pending_opening_params_requests .lock() .expect("lock") @@ -237,7 +252,7 @@ where { let response = LSPS1OpeningParamsResponse { supported_options }; - match sender.send(response) { + match request.sender.send(response) { Ok(()) => (), Err(_) => { log_error!( @@ -279,7 +294,7 @@ where .iter() .any(|n| n.node_id == counterparty_node_id) { - if let Some(sender) = + if let Some(request) = self.pending_create_order_requests.lock().expect("lock").remove(&request_id) { let response = LSPS1OrderStatus { @@ -290,7 +305,7 @@ where counterparty_node_id, }; - match sender.send(response) { + match request.sender.send(response) { Ok(()) => (), Err(_) => { log_error!( @@ -329,7 +344,7 @@ where .iter() .any(|n| n.node_id == counterparty_node_id) { - if let Some(sender) = self + if let Some(request) = self .pending_check_order_status_requests .lock() .expect("lock") @@ -343,7 +358,7 @@ where counterparty_node_id, }; - match sender.send(response) { + match request.sender.send(response) { Ok(()) => (), Err(_) => { log_error!( diff --git a/src/liquidity/client/lsps2.rs b/src/liquidity/client/lsps2.rs index 3033f8d827..4e6163e21b 100644 --- a/src/liquidity/client/lsps2.rs +++ b/src/liquidity/client/lsps2.rs @@ -26,8 +26,8 @@ use tokio::task::JoinSet; use crate::connection::ConnectionManager; use crate::liquidity::{ - select_all_lsps_for_protocol, select_lsps_for_protocol, LspConfig, LspNode, - LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, + select_all_lsps_for_protocol, select_lsps_for_protocol, LspConfig, LspNode, PendingRequest, + PendingRequestGuard, LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, }; use crate::logger::{log_debug, log_error, log_info, LdkLogger}; use crate::payment::store::LSPS2Parameters; @@ -41,9 +41,9 @@ where { pub(crate) lsp_nodes: Arc>>, pub(crate) pending_lsps2_fee_requests: - Mutex>>, + Mutex>>, pub(crate) pending_buy_requests: - Mutex>>, + Mutex>>, pub(crate) channel_manager: Arc, pub(crate) keys_manager: Arc, pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, @@ -262,13 +262,18 @@ where })?; let (fee_request_sender, fee_request_receiver) = oneshot::channel(); - { + let _pending_request = { let mut pending_fee_requests_lock = self.pending_lsps2_fee_requests.lock().expect("lock"); let request_id = client_handler.request_opening_params(lsps2_node.node_id, lsps2_node.token.clone()); - pending_fee_requests_lock.insert(request_id, fee_request_sender); - } + PendingRequestGuard::insert( + &self.pending_lsps2_fee_requests, + &mut pending_fee_requests_lock, + request_id, + fee_request_sender, + ) + }; tokio::time::timeout( Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), @@ -298,7 +303,7 @@ where })?; let (buy_request_sender, buy_request_receiver) = oneshot::channel(); - { + let _pending_request = { let mut pending_buy_requests_lock = self.pending_buy_requests.lock().expect("lock"); let request_id = client_handler .select_opening_params(lsps2_node.node_id, amount_msat, opening_fee_params) @@ -310,8 +315,13 @@ where ); Error::LiquidityRequestFailed })?; - pending_buy_requests_lock.insert(request_id, buy_request_sender); - } + PendingRequestGuard::insert( + &self.pending_buy_requests, + &mut pending_buy_requests_lock, + request_id, + buy_request_sender, + ) + }; let buy_response = tokio::time::timeout( Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), @@ -428,12 +438,12 @@ where .iter() .any(|n| n.node_id == counterparty_node_id) { - if let Some(sender) = + if let Some(request) = self.pending_lsps2_fee_requests.lock().expect("lock").remove(&request_id) { let response = LSPS2FeeResponse { opening_fee_params_menu }; - match sender.send(response) { + match request.sender.send(response) { Ok(()) => (), Err(_) => { log_error!( @@ -474,12 +484,12 @@ where .iter() .any(|n| n.node_id == counterparty_node_id) { - if let Some(sender) = + if let Some(request) = self.pending_buy_requests.lock().expect("lock").remove(&request_id) { let response = LSPS2BuyResponse { intercept_scid, cltv_expiry_delta }; - match sender.send(response) { + match request.sender.send(response) { Ok(()) => (), Err(_) => { log_error!( diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index 0eddde1ae8..ffc1f878bc 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -10,8 +10,8 @@ pub(crate) mod client; pub(crate) mod service; -use std::collections::hash_map::Entry; use std::collections::HashMap; +use std::hash::Hash; use std::ops::Deref; use std::sync::{Arc, Mutex, RwLock}; use std::time::Duration; @@ -42,6 +42,45 @@ use crate::{Config, Error}; const LIQUIDITY_REQUEST_TIMEOUT_SECS: u64 = 5; const LSPS_DISCOVERY_WAIT_TIMEOUT_SECS: u64 = 10; +pub(crate) struct PendingRequest { + token: Arc<()>, + pub(crate) sender: oneshot::Sender, + followers: Vec>, +} + +pub(crate) struct PendingRequestGuard<'a, K: Clone + Eq + Hash, T> { + pending_requests: &'a Mutex>>, + request_key: K, + token: Arc<()>, +} + +impl<'a, K: Clone + Eq + Hash, T> PendingRequestGuard<'a, K, T> { + pub(crate) fn insert( + pending_requests: &'a Mutex>>, + pending_requests_lock: &mut HashMap>, request_key: K, + sender: oneshot::Sender, + ) -> Self { + let token = Arc::new(()); + pending_requests_lock.insert( + request_key.clone(), + PendingRequest { token: Arc::clone(&token), sender, followers: Vec::new() }, + ); + Self { pending_requests, request_key, token } + } +} + +impl Drop for PendingRequestGuard<'_, K, T> { + fn drop(&mut self) { + let mut pending_requests = self.pending_requests.lock().expect("lock"); + if pending_requests + .get(&self.request_key) + .is_some_and(|request| Arc::ptr_eq(&request.token, &self.token)) + { + pending_requests.remove(&self.request_key); + } + } +} + fn select_lsps_for_protocol( lsp_nodes: &Arc>>, protocol: u16, override_node_id: Option<&PublicKey>, ) -> Option { @@ -345,7 +384,7 @@ where lsps1_client: Arc>, lsps2_client: Arc>, lsps2_service: Arc>, - pending_lsps0_discovery: Mutex>>>>, + pending_lsps0_discovery: Mutex>>>, discovery_done_tx: tokio::sync::watch::Sender, discovery_done_rx: tokio::sync::watch::Receiver, liquidity_manager: Arc, @@ -383,13 +422,14 @@ where protocols, }) => { if self.is_lsps_node(&counterparty_node_id) { - if let Some(senders) = self + if let Some(request) = self .pending_lsps0_discovery .lock() .expect("lock") .remove(&counterparty_node_id) { - for sender in senders { + let _ = request.sender.send(protocols.clone()); + for sender in request.followers { let _ = sender.send(protocols.clone()); } } else { @@ -431,23 +471,25 @@ where let lsps0_handler = self.liquidity_manager.lsps0_client_handler(); let (sender, receiver) = oneshot::channel(); - let issued_request = { + let _pending_request = { let mut pending_discovery = self.pending_lsps0_discovery.lock().expect("lock"); - match pending_discovery.entry(*node_id) { - Entry::Occupied(mut e) => { - e.get_mut().push(sender); - false - }, - Entry::Vacant(v) => { - v.insert(vec![sender]); - lsps0_handler.list_protocols(node_id); - true - }, + if let Some(request) = pending_discovery.get_mut(node_id) { + request.followers.push(sender); + None + } else { + let request_guard = PendingRequestGuard::insert( + &self.pending_lsps0_discovery, + &mut pending_discovery, + *node_id, + sender, + ); + lsps0_handler.list_protocols(node_id); + Some(request_guard) } }; - // Only the request that issued the discovery may remove the entry; a follower removing it - // would drop the in-flight request and all other waiters. + // Only the request that issued the discovery holds the guard. If it is abandoned, dropping + // the pending request also wakes all followers without risking removal of a newer request. let protocols = tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) .await @@ -458,9 +500,6 @@ where node_id, e ); - if issued_request { - self.pending_lsps0_discovery.lock().expect("lock").remove(node_id); - } Error::LiquidityRequestFailed })? .map_err(|e| { @@ -470,9 +509,6 @@ where node_id, e ); - if issued_request { - self.pending_lsps0_discovery.lock().expect("lock").remove(node_id); - } Error::LiquidityRequestFailed })?; @@ -540,3 +576,83 @@ where let _ = self.discovery_done_tx.send(true); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pending_request_guard_removes_abandoned_request() { + let pending_requests = Mutex::new(HashMap::new()); + let (sender, receiver) = oneshot::channel::<()>(); + let request_guard = { + let mut pending_requests_lock = pending_requests.lock().expect("lock"); + PendingRequestGuard::insert( + &pending_requests, + &mut pending_requests_lock, + "request".to_owned(), + sender, + ) + }; + let (follower_sender, mut follower_receiver) = oneshot::channel(); + pending_requests + .lock() + .expect("lock") + .get_mut("request") + .expect("request should be pending") + .followers + .push(follower_sender); + + drop(receiver); + drop(request_guard); + + assert!( + pending_requests.lock().expect("lock").is_empty(), + "abandoned pending request should be removed" + ); + assert!( + matches!(follower_receiver.try_recv(), Err(oneshot::error::TryRecvError::Closed)), + "abandoned pending request should wake its followers" + ); + } + + #[test] + fn pending_request_guard_preserves_replacement() { + let pending_requests = Mutex::new(HashMap::new()); + let (old_sender, old_receiver) = oneshot::channel::<()>(); + let old_request_guard = { + let mut pending_requests_lock = pending_requests.lock().expect("lock"); + PendingRequestGuard::insert( + &pending_requests, + &mut pending_requests_lock, + "request".to_owned(), + old_sender, + ) + }; + + pending_requests.lock().expect("lock").remove("request"); + drop(old_receiver); + + let (replacement_sender, replacement_receiver) = oneshot::channel::<()>(); + let replacement_request_guard = { + let mut pending_requests_lock = pending_requests.lock().expect("lock"); + PendingRequestGuard::insert( + &pending_requests, + &mut pending_requests_lock, + "request".to_owned(), + replacement_sender, + ) + }; + + drop(old_request_guard); + assert_eq!( + pending_requests.lock().expect("lock").len(), + 1, + "an older guard should not remove a replacement request" + ); + + drop(replacement_receiver); + drop(replacement_request_guard); + assert!(pending_requests.lock().expect("lock").is_empty()); + } +} From 9c1dbd22d08aed45f6562b89e001c23242cb4880 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 10 Aug 2026 12:06:24 +0200 Subject: [PATCH 06/10] f Tolerate late liquidity responses Treat responses for expired requests as expected so delayed LSP messages do not panic the liquidity event loop. Co-Authored-By: HAL 9000 --- src/liquidity/client/lsps1.rs | 12 ------------ src/liquidity/client/lsps2.rs | 8 -------- 2 files changed, 20 deletions(-) diff --git a/src/liquidity/client/lsps1.rs b/src/liquidity/client/lsps1.rs index ef6ceea638..30c1b0943c 100644 --- a/src/liquidity/client/lsps1.rs +++ b/src/liquidity/client/lsps1.rs @@ -263,10 +263,6 @@ where }, } } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); log_error!( self.logger, "Received response from liquidity service for unknown request." @@ -316,10 +312,6 @@ where }, } } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); log_error!( self.logger, "Received response from liquidity service for unknown request." @@ -369,10 +361,6 @@ where }, } } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); log_error!( self.logger, "Received response from liquidity service for unknown request." diff --git a/src/liquidity/client/lsps2.rs b/src/liquidity/client/lsps2.rs index 4e6163e21b..0bff733447 100644 --- a/src/liquidity/client/lsps2.rs +++ b/src/liquidity/client/lsps2.rs @@ -454,10 +454,6 @@ where }, } } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); log_error!( self.logger, "Received response from liquidity service for unknown request." @@ -500,10 +496,6 @@ where }, } } else { - debug_assert!( - false, - "Received response from liquidity service for unknown request." - ); log_error!( self.logger, "Received response from liquidity service for unknown request." From aeaec6db963c586e6758c3d14788e3ca138da671 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 20 Jul 2026 11:13:55 +0200 Subject: [PATCH 07/10] Clean up cancelled connection attempts Clear per-peer connection state when the leading task is cancelled so later callers can retry and existing subscribers do not hang. Co-Authored-By: HAL 9000 --- src/connection.rs | 87 +++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 73 insertions(+), 14 deletions(-) diff --git a/src/connection.rs b/src/connection.rs index 88135e841e..dda90d0ef6 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -18,12 +18,44 @@ use crate::logger::{log_debug, log_error, log_info, LdkLogger}; use crate::types::{KeysManager, PeerManager}; use crate::Error; +type PendingConnections = + Mutex>>>>; + +struct PendingConnectionGuard<'a> { + pending_connections: &'a PendingConnections, + node_id: PublicKey, + active: bool, +} + +impl<'a> PendingConnectionGuard<'a> { + fn new(pending_connections: &'a PendingConnections, node_id: PublicKey) -> Self { + Self { pending_connections, node_id, active: true } + } + + fn disarm(&mut self) { + self.active = false; + } +} + +impl Drop for PendingConnectionGuard<'_> { + fn drop(&mut self) { + if !self.active { + return; + } + let mut pending_connections = self.pending_connections.lock().expect("lock"); + if let Some(subscribers) = pending_connections.remove(&self.node_id) { + for subscriber in subscribers { + let _ = subscriber.send(Err(Error::ConnectionFailed)); + } + } + } +} + pub(crate) struct ConnectionManager where L::Target: LdkLogger, { - pending_connections: - Mutex>>>>, + pending_connections: PendingConnections, peer_manager: Arc, tor_proxy_config: Option, keys_manager: Arc, @@ -60,26 +92,29 @@ where pub(crate) async fn do_connect_peer( &self, node_id: PublicKey, addr: SocketAddress, ) -> Result<(), Error> { + // If another task is already connecting, subscribe to its result instead of starting a + // duplicate attempt. + if let Some(pending_connection_ready_receiver) = + self.register_or_subscribe_pending_connection(&node_id) + { + return pending_connection_ready_receiver.await.map_err(|e| { + debug_assert!(false, "Failed to receive connection result: {:?}", e); + log_error!(self.logger, "Failed to receive connection result: {:?}", e); + Error::ConnectionFailed + })?; + } + + let mut pending_connection = + PendingConnectionGuard::new(&self.pending_connections, node_id); let res = self.do_connect_peer_internal(node_id, addr).await; self.propagate_result_to_subscribers(&node_id, res); + pending_connection.disarm(); res } async fn do_connect_peer_internal( &self, node_id: PublicKey, addr: SocketAddress, ) -> Result<(), Error> { - // First, we check if there is already an outbound connection in flight, if so, we just - // await on the corresponding watch channel. The task driving the connection future will - // send us the result.. - let pending_ready_receiver_opt = self.register_or_subscribe_pending_connection(&node_id); - if let Some(pending_connection_ready_receiver) = pending_ready_receiver_opt { - return pending_connection_ready_receiver.await.map_err(|e| { - debug_assert!(false, "Failed to receive connection result: {:?}", e); - log_error!(self.logger, "Failed to receive connection result: {:?}", e); - Error::ConnectionFailed - })?; - } - log_info!(self.logger, "Connecting to peer: {}@{}", node_id, addr); match addr { @@ -246,6 +281,7 @@ where match pending_connections_lock.entry(*node_id) { hash_map::Entry::Occupied(mut entry) => { let (tx, rx) = tokio::sync::oneshot::channel(); + entry.get_mut().retain(|subscriber| !subscriber.is_closed()); entry.get_mut().push(tx); Some(rx) }, @@ -277,3 +313,26 @@ where } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pending_connection_guard_notifies_subscribers_when_abandoned() { + let node_id: PublicKey = + "0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798".parse().unwrap(); + let pending_connections = Mutex::new(HashMap::new()); + let (sender, mut receiver) = tokio::sync::oneshot::channel(); + pending_connections.lock().expect("lock").insert(node_id, vec![sender]); + + let connection_guard = PendingConnectionGuard::new(&pending_connections, node_id); + drop(connection_guard); + + assert!( + pending_connections.lock().expect("lock").is_empty(), + "abandoned connection attempt should remove pending state" + ); + assert_eq!(receiver.try_recv(), Ok(Err(Error::ConnectionFailed))); + } +} From 05a6c5a49168c50ce2d81e88a23ba4057fbd7537 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 27 Jul 2026 15:38:37 +0200 Subject: [PATCH 08/10] f Avoid panics for cancelled subscribers Connection subscribers may cancel before a shared attempt finishes. Treat the failed send as expected instead of panicking in debug builds. Co-Authored-By: HAL 9000 --- src/connection.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/connection.rs b/src/connection.rs index dda90d0ef6..ccb6f98464 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -298,11 +298,6 @@ where if let Some(connection_ready_senders) = pending_connections_lock.remove(node_id) { for sender in connection_ready_senders { let _ = sender.send(res).map_err(|e| { - debug_assert!( - false, - "Failed to send connection result to subscribers: {:?}", - e - ); log_error!( self.logger, "Failed to send connection result to subscribers: {:?}", From 7d60fd5be99c003b66c22838dae2548a7f2a2dcf Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 20 Jul 2026 11:33:18 +0200 Subject: [PATCH 09/10] Reset cancelled wallet sync state Restore wallet sync status and notify waiting callers when the task performing a sync is cancelled, allowing later sync attempts to run. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 14 ++++++---- src/chain/electrum.rs | 13 +++++---- src/chain/esplora.rs | 13 +++++---- src/chain/mod.rs | 64 ++++++++++++++++++++++++++++++++++++------- 4 files changed, 77 insertions(+), 27 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index f7589671b8..79f73f978a 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -31,7 +31,7 @@ use lightning_block_sync::{ }; use serde::Serialize; -use super::WalletSyncStatus; +use super::{WalletSyncGuard, WalletSyncStatus}; use crate::config::{ BitcoindRestClientConfig, Config, DEFAULT_FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, DEFAULT_TX_BROADCAST_TIMEOUT_SECS, @@ -160,12 +160,14 @@ impl BitcoindChainSource { ) { // First register for the wallet polling status to make sure `Node::sync_wallets` calls // wait on the result before proceeding. - { + let initial_sync_guard = { let mut status_lock = self.wallet_polling_status.lock().expect("lock"); if status_lock.register_or_subscribe_pending_sync().is_some() { debug_assert!(false, "Sync already in progress. This should never happen."); + return; } - } + WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed) + }; log_info!( self.logger, @@ -302,7 +304,7 @@ impl BitcoindChainSource { } // Now propagate the initial result to unblock waiting subscribers. - self.wallet_polling_status.lock().expect("lock").propagate_result_to_subscribers(Ok(())); + initial_sync_guard.complete(Ok(())); let mut chain_polling_interval = tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS)); @@ -413,6 +415,8 @@ impl BitcoindChainSource { Error::WalletOperationFailed })?; } + let sync_guard = + WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed); let res = self .poll_and_update_listeners_inner( @@ -423,7 +427,7 @@ impl BitcoindChainSource { ) .await; - self.wallet_polling_status.lock().expect("lock").propagate_result_to_subscribers(res); + sync_guard.complete(res); res } diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 59fa23a6ca..fbbebd7606 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -25,7 +25,7 @@ use lightning::chain::{Confirm, Filter, WatchedOutput}; use lightning::util::ser::Writeable; use lightning_transaction_sync::ElectrumSyncClient; -use super::WalletSyncStatus; +use super::{WalletSyncGuard, WalletSyncStatus}; use crate::config::{ clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, @@ -113,10 +113,12 @@ impl ElectrumChainSource { Error::WalletOperationFailed })?; } + let sync_guard = + WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed); let res = self.sync_onchain_wallet_inner(onchain_wallet).await; - self.onchain_wallet_sync_status.lock().expect("lock").propagate_result_to_subscribers(res); + sync_guard.complete(res); res } @@ -223,14 +225,13 @@ impl ElectrumChainSource { Error::TxSyncFailed })?; } + let sync_guard = + WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::TxSyncFailed); let res = self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await; - self.lightning_wallet_sync_status - .lock() - .expect("lock") - .propagate_result_to_subscribers(res); + sync_guard.complete(res); res } diff --git a/src/chain/esplora.rs b/src/chain/esplora.rs index f4aaea4c9b..1c13f141fb 100644 --- a/src/chain/esplora.rs +++ b/src/chain/esplora.rs @@ -18,7 +18,7 @@ use lightning::chain::{Confirm, Filter, WatchedOutput}; use lightning::util::ser::Writeable; use lightning_transaction_sync::EsploraSyncClient; -use super::WalletSyncStatus; +use super::{WalletSyncGuard, WalletSyncStatus}; use crate::config::{ clamp_full_scan_stop_gap, Config, EsploraSyncConfig, BDK_CLIENT_CONCURRENCY, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, @@ -133,10 +133,12 @@ impl EsploraChainSource { Error::WalletOperationFailed })?; } + let sync_guard = + WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed); let res = self.sync_onchain_wallet_inner(onchain_wallet).await; - self.onchain_wallet_sync_status.lock().expect("lock").propagate_result_to_subscribers(res); + sync_guard.complete(res); res } @@ -283,14 +285,13 @@ impl EsploraChainSource { Error::WalletOperationFailed })?; } + let sync_guard = + WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::WalletOperationFailed); let res = self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await; - self.lightning_wallet_sync_status - .lock() - .expect("lock") - .propagate_result_to_subscribers(res); + sync_guard.complete(res); res } diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 0f96c409f8..4aeb64704c 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -65,6 +65,34 @@ pub(crate) enum WalletSyncStatus { InProgress { subscribers: tokio::sync::broadcast::Sender> }, } +pub(crate) struct WalletSyncGuard<'a> { + status: &'a Mutex, + cancellation_error: Error, + active: bool, +} + +impl<'a> WalletSyncGuard<'a> { + pub(crate) fn new(status: &'a Mutex, cancellation_error: Error) -> Self { + Self { status, cancellation_error, active: true } + } + + pub(crate) fn complete(mut self, res: Result<(), Error>) { + self.status.lock().expect("lock").propagate_result_to_subscribers(res); + self.active = false; + } +} + +impl Drop for WalletSyncGuard<'_> { + fn drop(&mut self) { + if self.active { + self.status + .lock() + .expect("lock") + .propagate_result_to_subscribers(Err(self.cancellation_error)); + } + } +} + impl WalletSyncStatus { fn register_or_subscribe_pending_sync( &mut self, @@ -95,16 +123,7 @@ impl WalletSyncStatus { WalletSyncStatus::InProgress { subscribers } => { // A sync is in-progress, we notify subscribers. if subscribers.receiver_count() > 0 { - match subscribers.send(res) { - Ok(_) => (), - Err(e) => { - debug_assert!( - false, - "Failed to send wallet sync result to subscribers: {:?}", - e - ); - }, - } + let _ = subscribers.send(res); } *self = WalletSyncStatus::Completed; }, @@ -561,3 +580,28 @@ impl Filter for ChainSource { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn wallet_sync_guard_resets_abandoned_sync() { + let status = Mutex::new(WalletSyncStatus::Completed); + assert!(status.lock().expect("lock").register_or_subscribe_pending_sync().is_none()); + let sync_guard = WalletSyncGuard::new(&status, Error::WalletOperationFailed); + let mut subscriber = status + .lock() + .expect("lock") + .register_or_subscribe_pending_sync() + .expect("sync subscriber"); + + drop(sync_guard); + + assert!( + matches!(*status.lock().expect("lock"), WalletSyncStatus::Completed), + "abandoned wallet sync should reset its status" + ); + assert_eq!(subscriber.try_recv(), Ok(Err(Error::WalletOperationFailed))); + } +} From 2c1682e1e197492a2cb7c1490b2186e3fd92c293 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Mon, 10 Aug 2026 12:08:04 +0200 Subject: [PATCH 10/10] f Keep bitcoind background sync alive Wait for an active manual sync and retry initial ownership so startup races do not permanently stop background polling. Co-Authored-By: HAL 9000 --- src/chain/bitcoind.rs | 72 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 63 insertions(+), 9 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 79f73f978a..200fa3d3a2 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -52,6 +52,31 @@ const CHAIN_POLLING_TIMEOUT_SECS: u64 = 10; type BitcoindSpvClient = SpvClient, BitcoindClient>, Arc>; +async fn acquire_initial_wallet_sync_guard<'a>( + wallet_polling_status: &'a Mutex, + stop_sync_receiver: &mut tokio::sync::watch::Receiver<()>, +) -> Option> { + loop { + let mut pending_sync = { + let mut status_lock = wallet_polling_status.lock().expect("lock"); + match status_lock.register_or_subscribe_pending_sync() { + Some(pending_sync) => pending_sync, + None => { + return Some(WalletSyncGuard::new( + wallet_polling_status, + Error::WalletOperationFailed, + )); + }, + } + }; + tokio::select! { + biased; + _ = stop_sync_receiver.changed() => return None, + _ = pending_sync.recv() => {}, + } + } +} + pub(super) struct BitcoindChainSource { api_client: Arc, spv_client: tokio::sync::Mutex>, @@ -160,13 +185,12 @@ impl BitcoindChainSource { ) { // First register for the wallet polling status to make sure `Node::sync_wallets` calls // wait on the result before proceeding. - let initial_sync_guard = { - let mut status_lock = self.wallet_polling_status.lock().expect("lock"); - if status_lock.register_or_subscribe_pending_sync().is_some() { - debug_assert!(false, "Sync already in progress. This should never happen."); - return; - } - WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed) + let Some(initial_sync_guard) = + acquire_initial_wallet_sync_guard(&self.wallet_polling_status, &mut stop_sync_receiver) + .await + else { + log_trace!(self.logger, "Stopping initial chain sync."); + return; }; log_info!( @@ -1592,6 +1616,9 @@ impl std::error::Error for BitcoindClientError {} #[cfg(test)] mod tests { + use std::sync::Mutex; + use std::time::Duration; + use bitcoin::hashes::Hash; use bitcoin::{FeeRate, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness}; use lightning_block_sync::http::JsonResponse; @@ -1601,9 +1628,36 @@ mod tests { use serde_json::json; use crate::chain::bitcoind::{ - FeeResponse, GetMempoolEntryResponse, GetRawMempoolResponse, GetRawTransactionResponse, - MempoolMinFeeResponse, + acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse, + GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse, }; + use crate::chain::{WalletSyncGuard, WalletSyncStatus}; + use crate::Error; + + #[tokio::test] + async fn initial_sync_waits_for_in_progress_sync() { + let status = Mutex::new(WalletSyncStatus::Completed); + assert!(status.lock().expect("lock").register_or_subscribe_pending_sync().is_none()); + let in_progress_guard = WalletSyncGuard::new(&status, Error::WalletOperationFailed); + let (_stop_sender, mut stop_receiver) = tokio::sync::watch::channel(()); + let mut acquire_guard = + Box::pin(acquire_initial_wallet_sync_guard(&status, &mut stop_receiver)); + + let early_result = + tokio::time::timeout(Duration::from_millis(10), acquire_guard.as_mut()).await; + assert!(early_result.is_err(), "background sync should wait for the active sync"); + + in_progress_guard.complete(Ok(())); + let acquired_guard = tokio::time::timeout(Duration::from_secs(1), acquire_guard) + .await + .expect("background sync should resume") + .expect("background sync should acquire the sync guard"); + assert!( + matches!(*status.lock().expect("lock"), WalletSyncStatus::InProgress { .. }), + "background sync should own the next sync" + ); + acquired_guard.complete(Ok(())); + } prop_compose! { fn arbitrary_witness()(