From e18c6cde20930ebf984a350ba2b773a586098614 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 13:45:23 -0500 Subject: [PATCH 01/10] Poll bitcoind every second in the CLN interop tests lightningd polls bitcoind every 30 seconds by default, so after the tests mine the blocks a channel needs it can take CLN half a minute to notice them, plus the time to walk them, before it sends `channel_ready`. The tests wait 60 seconds for the resulting `ChannelReady` event, which left a few seconds of margin on a busy CI runner and timed out in `test_basic_channel_cycle`. The same scenario takes 8 seconds against LND and 12 against Eclair, which get notified of blocks over ZMQ. The poll interval is a developer option, and developer mode turns deprecated APIs off, which drops the `bech32` address from `newaddr` and the `tx`/`txid` fields from `close` that the tests read, so deprecated APIs are explicitly allowed again. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/docker/docker-compose-cln.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/docker/docker-compose-cln.yml b/tests/docker/docker-compose-cln.yml index 0f226f304f..f739bfe800 100644 --- a/tests/docker/docker-compose-cln.yml +++ b/tests/docker/docker-compose-cln.yml @@ -68,6 +68,12 @@ services: "--bitcoin-rpcpassword=pass", "--regtest", "--experimental-splicing", + # lightningd polls bitcoind every 30 seconds unless told otherwise, and the override is a + # developer option. Developer mode also turns deprecated APIs off, which changes the + # `newaddr` and `close` responses the tests read, so they are turned back on. + "--developer", + "--allow-deprecated-apis=true", + "--dev-bitcoind-poll=1", ] ports: - "19846:19846" From 6cbd3701059981f3c1d71efff9c30470b16f0f29 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 13:45:24 -0500 Subject: [PATCH 02/10] Wait for the external node's chain view before expecting ChannelReady `channel_ready` only arrives once the external node has seen the blocks the test mined, but the scenarios started the 60-second event wait right after mining, so an external node that lags its chain backend ate into that budget and a timeout there read as a channel problem. The setup already waits for the peer to reach Bitcoin Core's height before funding; do the same after mining the blocks a channel open or a splice lock needs, so the event wait measures the channel exchange alone and a lagging peer fails with a message that says so. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/common/scenarios/channel.rs | 1 + tests/common/scenarios/mod.rs | 15 +++++++++++++-- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/common/scenarios/channel.rs b/tests/common/scenarios/channel.rs index 10a8aaf0cc..dde7133185 100644 --- a/tests/common/scenarios/channel.rs +++ b/tests/common/scenarios/channel.rs @@ -30,6 +30,7 @@ pub(crate) async fn open_channel_to_external( let funding_txo = expect_channel_pending_event!(node, ext_node_id); super::super::wait_for_tx(electrs, funding_txo.txid).await; generate_blocks_and_wait(bitcoind, electrs, 10).await; + super::wait_for_peer_block_sync(peer, bitcoind).await; super::sync_wallets_with_retry(node).await; let user_channel_id = expect_channel_ready_event!(node, ext_node_id); diff --git a/tests/common/scenarios/mod.rs b/tests/common/scenarios/mod.rs index addc34681d..2c4a4521ce 100644 --- a/tests/common/scenarios/mod.rs +++ b/tests/common/scenarios/mod.rs @@ -52,6 +52,17 @@ where unreachable!() } +/// Waits until the peer has seen every block Bitcoin Core has. An implementation that polls +/// its chain backend (CLN every `bitcoind-poll` seconds) sees freshly mined blocks late, so a +/// message it only sends once it has, such as `channel_ready`, must not be waited for on the +/// event timeout alone. +pub(crate) async fn wait_for_peer_block_sync( + peer: &(impl ExternalNode + ?Sized), bitcoind: &BitcoindClient, +) { + let chain_height: u64 = bitcoind.get_blockchain_info().unwrap().blocks.try_into().unwrap(); + peer.wait_for_block_sync(chain_height).await.unwrap(); +} + /// Sync wallets, retrying on `WalletOperationTimeout`. pub(crate) async fn sync_wallets_with_retry(node: &Node) { for attempt in 0..3 { @@ -123,8 +134,7 @@ pub(crate) async fn setup_interop_test( generate_blocks_and_wait(bitcoind, electrs, 1).await; // Block until the peer indexes the funding tx, else channel opens time out. - let chain_height: u64 = bitcoind.get_blockchain_info().unwrap().blocks.try_into().unwrap(); - peer.wait_for_block_sync(chain_height).await.unwrap(); + wait_for_peer_block_sync(peer, bitcoind).await; sync_wallets_with_retry(node).await; @@ -243,6 +253,7 @@ pub(crate) async fn splice_in_scenario( let splice_txo = expect_splice_negotiated_event!(node, ext_node_id); wait_for_tx(electrs, splice_txo.txid).await; generate_blocks_and_wait(bitcoind, electrs, 6).await; + wait_for_peer_block_sync(peer, bitcoind).await; sync_wallets_with_retry(node).await; expect_channel_ready_event!(node, ext_node_id); From 653c335d5a342b049a834bdbd8eb1dfbbc34acf5 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 13:45:25 -0500 Subject: [PATCH 03/10] Give the test harness's polls a time budget instead of a try count `exponential_backoff_poll` gave up after 20 tries, about nine seconds in all once the backoff had reached its cap. That is not enough for electrs to index a hundred freshly generated blocks on a busy CI runner: `postgres_node_restart` died in the premine wait and `reorg_test` after generating 139 blocks, both with "Reached max tries". Poll for up to 60 seconds instead; a condition that already holds still returns at once, so passing runs are unaffected. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/common/mod.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index bb1f366b1c..d7c6966bbd 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -76,6 +76,9 @@ pub(crate) use in_memory_store::InMemoryStore; /// Shared timeout (in seconds) for waiting on LDK events and external node operations. pub(crate) const INTEROP_TIMEOUT_SECS: u64 = 60; +/// How long `exponential_backoff_poll` keeps polling before it gives up. +pub(crate) const POLL_TIMEOUT_SECS: u64 = 60; + macro_rules! expect_event { ($node:expr, $event_type:ident) => {{ let event = tokio::time::timeout( @@ -1012,8 +1015,11 @@ pub(crate) async fn exponential_backoff_poll(mut poll: F) -> T where F: FnMut() -> Option, { + // Budgeted by time rather than by tries: a try count gave about nine seconds in all, which a + // busy CI runner exceeds just indexing a hundred freshly generated blocks. A condition that + // already holds returns at once either way. + let deadline = tokio::time::Instant::now() + Duration::from_secs(POLL_TIMEOUT_SECS); let mut delay = Duration::from_millis(64); - let mut tries = 0; loop { match poll() { Some(data) => break data, @@ -1023,8 +1029,7 @@ where None => {}, } - assert!(tries < 20, "Reached max tries."); - tries += 1; + assert!(tokio::time::Instant::now() < deadline, "Timed out after {}s.", POLL_TIMEOUT_SECS); tokio::time::sleep(delay).await; } } From 7e67c5eeebef6fd42c505bb811cc5d754d3aae1b Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 15:03:33 -0500 Subject: [PATCH 04/10] Wait for the channel to be reestablished before closing it a_close_is_classified_for_a_channel_whose_facts_were_never_recorded reconnects a restarted node to its peer and immediately asks for a cooperative close. Connecting returns once the peer handshake is done, but the close is refused until channel_reestablish has been exchanged over the new connection, which takes another message round trip. On a slow runner the close came first and the test failed with ChannelClosingFailed. Poll the channel until it reports usable, which includes having been reestablished, before closing it. Another test already does this after a reconnect. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/integration_tests_rust.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 080fb8c359..01693f6c17 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -713,6 +713,9 @@ async fn a_close_is_classified_for_a_channel_whose_facts_were_never_recorded() { let node_addr_b = node_b.listening_addresses().unwrap().first().unwrap().clone(); node_a.connect(node_b.node_id(), node_addr_b, false).unwrap(); + // Connecting returns once the handshake is done, but a cooperative close is refused until the + // channel has been reestablished over the new connection. + wait_for_channel_ready_to_send(&node_a, &node_b, 0).await; let user_channel_id_a = node_a.list_channels().first().unwrap().user_channel_id; node_a.close_channel(&user_channel_id_a, node_b.node_id()).unwrap(); expect_event!(node_a, ChannelClosed); From 9d22ba397e8e31d17dd2ae5932415893ba7359a2 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 18:09:58 -0500 Subject: [PATCH 05/10] Finish every TestSyncStore operation once it has started TestSyncStore mirrors each write and remove to a filesystem store, a SQLite store and an in-memory store, and asserts on every access that they agree. The mirrors were applied one after another with awaits in between, so a caller that dropped the future partway left the stores disagreeing: the filesystem and SQLite writes had already been handed to blocking threads and landed, while the in-memory one never ran. The node does exactly that when it stops: its cancellable background tasks, several of which persist the node metrics, are dropped wherever they are suspended. The next node built on the same store then failed its first read with a consistency assertion in read_node_metrics, which is how a_sweep_confirmed_while_down_is_classified_for_a_channel_whose_facts_were_never_recorded failed in about one run out of six locally. Run each operation on a runtime the store owns and only await its completion from the caller, so dropping the caller's future no longer interrupts it. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/common/mod.rs | 101 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index d7c6966bbd..bc6d726b3c 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -2032,6 +2032,10 @@ impl PaginatedKVStore for TestSyncStore { struct TestSyncStoreInner { serializer: tokio::sync::RwLock<()>, + // Carries every write and remove through to all stores. A caller may drop its future midway, + // as the node does with a cancellable background task when it stops, and the stores would + // otherwise end up disagreeing with each other. + mirror_runtime: Option, test_store: InMemoryStore, fs_store: FilesystemStore, #[cfg(feature = "storage-sqlite")] @@ -2056,8 +2060,15 @@ impl TestSyncStoreInner { ) .unwrap(); let test_store = InMemoryStore::new(); + let mirror_runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .worker_threads(1) + .thread_name("test-sync-store") + .build() + .unwrap(); Self { serializer, + mirror_runtime: Some(mirror_runtime), fs_store, #[cfg(feature = "storage-sqlite")] sqlite_store, @@ -2177,6 +2188,20 @@ impl TestSyncStoreInner { } async fn write_internal_async( + self: &Arc, primary_namespace: &str, secondary_namespace: &str, key: &str, + buf: Vec, + ) -> lightning::io::Result<()> { + let inner = Arc::clone(self); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + let write = self.mirror_runtime().spawn(async move { + inner.do_write_async(&primary_namespace, &secondary_namespace, &key, buf).await + }); + write.await.unwrap_or_else(|e| resume_mirror_panic(e)) + } + + async fn do_write_async( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, ) -> lightning::io::Result<()> { let _guard = self.serializer.write().await; @@ -2229,6 +2254,19 @@ impl TestSyncStoreInner { } async fn remove_internal_async( + self: &Arc, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> lightning::io::Result<()> { + let inner = Arc::clone(self); + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + let remove = self.mirror_runtime().spawn(async move { + inner.do_remove_async(&primary_namespace, &secondary_namespace, &key, lazy).await + }); + remove.await.unwrap_or_else(|e| resume_mirror_panic(e)) + } + + async fn do_remove_async( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, ) -> lightning::io::Result<()> { let _guard = self.serializer.write().await; @@ -2266,6 +2304,69 @@ impl TestSyncStoreInner { } } +impl TestSyncStoreInner { + fn mirror_runtime(&self) -> &tokio::runtime::Handle { + self.mirror_runtime.as_ref().expect("the runtime lives as long as the store").handle() + } +} + +impl Drop for TestSyncStoreInner { + fn drop(&mut self) { + // Dropping a runtime from within another runtime's context is not allowed; shutting it + // down in the background is. + if let Some(runtime) = self.mirror_runtime.take() { + runtime.shutdown_background(); + } + } +} + +/// The stores' consistency assertions run on the mirror runtime; a failed one must still fail +/// the test that issued the operation. +fn resume_mirror_panic(e: tokio::task::JoinError) -> T { + match e.try_into_panic() { + Ok(payload) => std::panic::resume_unwind(payload), + Err(e) => panic!("the mirrored store operation did not finish: {}", e), + } +} + +#[cfg(test)] +mod test_sync_store_tests { + use super::*; + + // When the node stops it drops the futures of its cancellable background tasks wherever they + // are suspended, which can be partway through a store operation. Every store must still see + // the operation, or the next read trips the consistency assertions. + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + async fn an_operation_whose_future_is_dropped_still_reaches_every_store() { + let store = TestSyncStore::new(random_storage_path()); + let (ns, key, value) = ("ns", "key", vec![1, 2, 3]); + + // Polls the write once and then drops it. + let write = KVStore::write(&store, ns, "", key, value.clone()); + let _ = tokio::time::timeout(Duration::ZERO, write).await; + wait_until(|| async { + KVStore::read(&store, ns, "", key).await.ok() == Some(value.clone()) + }) + .await; + + let remove = KVStore::remove(&store, ns, "", key, false); + let _ = tokio::time::timeout(Duration::ZERO, remove).await; + wait_until(|| async { KVStore::read(&store, ns, "", key).await.is_err() }).await; + } + + async fn wait_until(mut condition: F) + where + F: FnMut() -> Fut, + Fut: Future, + { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + while !condition().await { + assert!(tokio::time::Instant::now() < deadline, "the store operation never landed"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } +} + /// The PostgreSQL connection string used by the Postgres-backed tests, overridable via the /// `TEST_POSTGRES_URL` environment variable. #[cfg(feature = "storage-postgres")] From 5c590438afceb734dd421c371255964eaee02d20 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 19:37:47 -0500 Subject: [PATCH 06/10] Sync to the current tip even when a sync is already running `Node::sync_wallets` returned as soon as a sync that was already in progress finished, taking that sync's result as its own. That sync took its view of the chain when it started, so the caller could get back a node that was still behind the tip. With a Bitcoin Core chain source this is easy to hit: blocks are processed one at a time, with the channel monitors, the sweeper and the on-chain wallet persisted for each, so a background poll that started while a batch of blocks was being mined keeps running long after mining is done. A manual sync arriving in the meantime piggybacked on it and returned with the rest of the batch still unprocessed. The reorg test then timed out on CI: the sweep it waited for could only be broadcast once the remaining blocks had been processed, which now happened inside the wait's time budget rather than in the sync call before it. A sync now waits for the one in progress and then runs its own, so it reflects the tip as of when it was called. A failed or cancelled sync in progress still fails the waiting caller as before. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- src/chain/bitcoind.rs | 76 +++++++++++++++++++++++++++++++++---------- src/chain/electrum.rs | 42 ++++++++---------------- src/chain/esplora.rs | 42 ++++++++---------------- src/chain/mod.rs | 28 +++++++++++++++- 4 files changed, 112 insertions(+), 76 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 2628294160..0329b8a334 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -33,7 +33,7 @@ use lightning_block_sync::{ }; use serde::Serialize; -use super::{ChainListener, WalletSyncGuard, WalletSyncStatus}; +use super::{acquire_wallet_sync_guard, ChainListener, WalletSyncGuard, WalletSyncStatus}; use crate::config::{ BitcoindRestClientConfig, Config, DEFAULT_FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, DEFAULT_TX_BROADCAST_TIMEOUT_SECS, @@ -448,21 +448,12 @@ impl BitcoindChainSource { &self, onchain_wallet: Arc, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, ) -> Result<(), Error> { - let receiver_res = { - let mut status_lock = self.wallet_polling_status.lock().expect("lock"); - status_lock.register_or_subscribe_pending_sync() - }; - - if let Some(mut sync_receiver) = receiver_res { - log_debug!(self.logger, "Sync in progress, skipping."); - return sync_receiver.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet polling result: {:?}", e); - log_error!(self.logger, "Failed to receive wallet polling result: {:?}", e); - Error::WalletOperationFailed - })?; - } - let sync_guard = - WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed); + let sync_guard = acquire_wallet_sync_guard( + &self.wallet_polling_status, + Error::WalletOperationFailed, + &self.logger, + ) + .await?; let res = self .poll_and_update_listeners_inner( @@ -1599,7 +1590,8 @@ mod tests { acquire_initial_wallet_sync_guard, FeeResponse, GetMempoolEntryResponse, GetRawMempoolResponse, GetRawTransactionResponse, MempoolMinFeeResponse, }; - use crate::chain::{WalletSyncGuard, WalletSyncStatus}; + use crate::chain::{acquire_wallet_sync_guard, WalletSyncGuard, WalletSyncStatus}; + use crate::logger::Logger; use crate::Error; #[tokio::test] @@ -1627,6 +1619,56 @@ mod tests { acquired_guard.complete(Ok(())); } + #[tokio::test] + async fn a_sync_waits_for_the_one_in_progress_and_then_runs_its_own() { + let logger = Logger::new_log_facade(); + 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 mut acquire_guard = + Box::pin(acquire_wallet_sync_guard(&status, Error::WalletOperationFailed, &logger)); + + let early_result = + tokio::time::timeout(Duration::from_millis(10), acquire_guard.as_mut()).await; + assert!(early_result.is_err(), "the caller should wait for the sync in progress"); + + in_progress_guard.complete(Ok(())); + let acquired_guard = tokio::time::timeout(Duration::from_secs(1), acquire_guard) + .await + .expect("the caller should resume once the sync in progress is done") + .expect("the caller should acquire the sync guard"); + assert!( + matches!(*status.lock().expect("lock"), WalletSyncStatus::InProgress { .. }), + "the caller should own the next sync rather than return" + ); + acquired_guard.complete(Ok(())); + assert!(matches!(*status.lock().expect("lock"), WalletSyncStatus::Completed)); + } + + #[tokio::test] + async fn a_sync_fails_when_the_one_in_progress_fails() { + let logger = Logger::new_log_facade(); + 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 mut acquire_guard = + Box::pin(acquire_wallet_sync_guard(&status, Error::WalletOperationFailed, &logger)); + + let early_result = + tokio::time::timeout(Duration::from_millis(10), acquire_guard.as_mut()).await; + assert!(early_result.is_err(), "the caller should wait for the sync in progress"); + + in_progress_guard.complete(Err(Error::TxSyncFailed)); + let result = tokio::time::timeout(Duration::from_secs(1), acquire_guard) + .await + .expect("the caller should resume once the sync in progress is done"); + assert_eq!(result.err(), Some(Error::TxSyncFailed)); + assert!( + matches!(*status.lock().expect("lock"), WalletSyncStatus::Completed), + "a failed caller should not hold the sync status" + ); + } + prop_compose! { fn arbitrary_witness()( witness_elements in vec(vec(any::(), 0..100), 0..20) diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 86025998e8..aaaaa74a36 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::{WalletSyncGuard, WalletSyncStatus}; +use super::{acquire_wallet_sync_guard, WalletSyncStatus}; use crate::config::{ clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, @@ -111,20 +111,12 @@ impl ElectrumChainSource { pub(crate) async fn sync_onchain_wallet( &self, onchain_wallet: Arc, ) -> Result<(), Error> { - let receiver_res = { - let mut status_lock = self.onchain_wallet_sync_status.lock().expect("lock"); - status_lock.register_or_subscribe_pending_sync() - }; - if let Some(mut sync_receiver) = receiver_res { - log_debug!(self.logger, "Sync in progress, skipping."); - return sync_receiver.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); - log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e); - Error::WalletOperationFailed - })?; - } - let sync_guard = - WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed); + let sync_guard = acquire_wallet_sync_guard( + &self.onchain_wallet_sync_status, + Error::WalletOperationFailed, + &self.logger, + ) + .await?; let res = self.sync_onchain_wallet_inner(onchain_wallet).await; @@ -223,20 +215,12 @@ impl ElectrumChainSource { &self, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, ) -> Result<(), Error> { - let receiver_res = { - let mut status_lock = self.lightning_wallet_sync_status.lock().expect("lock"); - status_lock.register_or_subscribe_pending_sync() - }; - if let Some(mut sync_receiver) = receiver_res { - log_debug!(self.logger, "Sync in progress, skipping."); - return sync_receiver.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); - log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e); - Error::TxSyncFailed - })?; - } - let sync_guard = - WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::TxSyncFailed); + let sync_guard = acquire_wallet_sync_guard( + &self.lightning_wallet_sync_status, + Error::TxSyncFailed, + &self.logger, + ) + .await?; let res = self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await; diff --git a/src/chain/esplora.rs b/src/chain/esplora.rs index 1c13f141fb..6d3762e548 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::{WalletSyncGuard, WalletSyncStatus}; +use super::{acquire_wallet_sync_guard, WalletSyncStatus}; use crate::config::{ clamp_full_scan_stop_gap, Config, EsploraSyncConfig, BDK_CLIENT_CONCURRENCY, MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, @@ -121,20 +121,12 @@ impl EsploraChainSource { pub(super) async fn sync_onchain_wallet( &self, onchain_wallet: Arc, ) -> Result<(), Error> { - let receiver_res = { - let mut status_lock = self.onchain_wallet_sync_status.lock().expect("lock"); - status_lock.register_or_subscribe_pending_sync() - }; - if let Some(mut sync_receiver) = receiver_res { - log_debug!(self.logger, "Sync in progress, skipping."); - return sync_receiver.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); - log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e); - Error::WalletOperationFailed - })?; - } - let sync_guard = - WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed); + let sync_guard = acquire_wallet_sync_guard( + &self.onchain_wallet_sync_status, + Error::WalletOperationFailed, + &self.logger, + ) + .await?; let res = self.sync_onchain_wallet_inner(onchain_wallet).await; @@ -273,20 +265,12 @@ impl EsploraChainSource { &self, channel_manager: Arc, chain_monitor: Arc, output_sweeper: Arc, ) -> Result<(), Error> { - let receiver_res = { - let mut status_lock = self.lightning_wallet_sync_status.lock().expect("lock"); - status_lock.register_or_subscribe_pending_sync() - }; - if let Some(mut sync_receiver) = receiver_res { - log_debug!(self.logger, "Sync in progress, skipping."); - return sync_receiver.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); - log_error!(self.logger, "Failed to receive wallet sync result: {:?}", e); - Error::WalletOperationFailed - })?; - } - let sync_guard = - WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::WalletOperationFailed); + let sync_guard = acquire_wallet_sync_guard( + &self.lightning_wallet_sync_status, + Error::WalletOperationFailed, + &self.logger, + ) + .await?; let res = self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await; diff --git a/src/chain/mod.rs b/src/chain/mod.rs index e1f7e2ad41..40527112d7 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -37,7 +37,7 @@ use crate::config::ElectrumSyncConfig; use crate::config::EsploraSyncConfig; use crate::config::{BackgroundSyncConfig, Config, WALLET_SYNC_INTERVAL_MINIMUM_SECS}; use crate::fee_estimator::OnchainFeeEstimator; -use crate::logger::{log_debug, log_info, log_trace, LdkLogger, Logger}; +use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger}; use crate::runtime::Runtime; use crate::tx_broadcaster::BroadcastPackage; use crate::types::{Broadcaster, ChainMonitor, ChannelManager, DynStore, Sweeper, Wallet}; @@ -107,6 +107,32 @@ impl Drop for WalletSyncGuard<'_> { } } +/// Waits for any sync that is already in progress and then registers the caller as the next one. +/// +/// A sync that is already running took its view of the chain when it started, so it might not +/// include blocks found since then. A caller that wants the current chain state therefore has to +/// sync again once it is done. If the sync in progress fails or is cancelled, the caller fails +/// with its error instead. +pub(crate) async fn acquire_wallet_sync_guard<'a>( + status: &'a Mutex, cancellation_error: Error, logger: &Logger, +) -> Result, Error> { + loop { + let mut pending_sync = { + let mut status_lock = status.lock().expect("lock"); + match status_lock.register_or_subscribe_pending_sync() { + Some(pending_sync) => pending_sync, + None => return Ok(WalletSyncGuard::new(status, cancellation_error)), + } + }; + log_debug!(logger, "Sync in progress, waiting for it to finish before syncing again."); + pending_sync.recv().await.map_err(|e| { + debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); + log_error!(logger, "Failed to receive wallet sync result: {:?}", e); + cancellation_error + })??; + } +} + impl WalletSyncStatus { fn register_or_subscribe_pending_sync( &mut self, From 5f3dc003e9485972d627d142d0367970a0c213cd Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 20 Oct 2026 17:51:14 -0500 Subject: [PATCH 07/10] Sync anyway when the sync in progress failed Since a caller runs its own sync after waiting for the one in progress, that sync's result no longer stands in for the caller's. Returning its error still failed the caller without it having synced at all. With background syncing on, a manual `sync_wallets` call that arrived while a background sync was failing returned that failure as its own. The caller now logs the failure and syncs anyway, so what it returns is the result of its own sync. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- src/chain/bitcoind.rs | 42 ++++++++++++++++++++---------------------- src/chain/mod.rs | 20 +++++++++++++------- 2 files changed, 33 insertions(+), 29 deletions(-) diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 0329b8a334..bf2312e39c 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -1619,8 +1619,12 @@ mod tests { acquired_guard.complete(Ok(())); } - #[tokio::test] - async fn a_sync_waits_for_the_one_in_progress_and_then_runs_its_own() { + /// Starts a sync, lets a second caller wait on it, ends the first one with + /// `in_progress_result` (`None` drops its guard, as a cancelled sync does) and checks that the + /// second caller then runs its own sync. + async fn assert_a_sync_runs_after_the_one_in_progress_ends( + in_progress_result: Option>, + ) { let logger = Logger::new_log_facade(); let status = Mutex::new(WalletSyncStatus::Completed); assert!(status.lock().expect("lock").register_or_subscribe_pending_sync().is_none()); @@ -1632,7 +1636,10 @@ mod tests { tokio::time::timeout(Duration::from_millis(10), acquire_guard.as_mut()).await; assert!(early_result.is_err(), "the caller should wait for the sync in progress"); - in_progress_guard.complete(Ok(())); + match in_progress_result { + Some(res) => in_progress_guard.complete(res), + None => drop(in_progress_guard), + } let acquired_guard = tokio::time::timeout(Duration::from_secs(1), acquire_guard) .await .expect("the caller should resume once the sync in progress is done") @@ -1646,27 +1653,18 @@ mod tests { } #[tokio::test] - async fn a_sync_fails_when_the_one_in_progress_fails() { - let logger = Logger::new_log_facade(); - 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 mut acquire_guard = - Box::pin(acquire_wallet_sync_guard(&status, Error::WalletOperationFailed, &logger)); + async fn a_sync_waits_for_the_one_in_progress_and_then_runs_its_own() { + assert_a_sync_runs_after_the_one_in_progress_ends(Some(Ok(()))).await; + } - let early_result = - tokio::time::timeout(Duration::from_millis(10), acquire_guard.as_mut()).await; - assert!(early_result.is_err(), "the caller should wait for the sync in progress"); + #[tokio::test] + async fn a_sync_runs_after_the_one_in_progress_fails() { + assert_a_sync_runs_after_the_one_in_progress_ends(Some(Err(Error::TxSyncFailed))).await; + } - in_progress_guard.complete(Err(Error::TxSyncFailed)); - let result = tokio::time::timeout(Duration::from_secs(1), acquire_guard) - .await - .expect("the caller should resume once the sync in progress is done"); - assert_eq!(result.err(), Some(Error::TxSyncFailed)); - assert!( - matches!(*status.lock().expect("lock"), WalletSyncStatus::Completed), - "a failed caller should not hold the sync status" - ); + #[tokio::test] + async fn a_sync_runs_after_the_one_in_progress_is_cancelled() { + assert_a_sync_runs_after_the_one_in_progress_ends(None).await; } prop_compose! { diff --git a/src/chain/mod.rs b/src/chain/mod.rs index 40527112d7..c360a06077 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -111,8 +111,8 @@ impl Drop for WalletSyncGuard<'_> { /// /// A sync that is already running took its view of the chain when it started, so it might not /// include blocks found since then. A caller that wants the current chain state therefore has to -/// sync again once it is done. If the sync in progress fails or is cancelled, the caller fails -/// with its error instead. +/// sync again once it is done. Whether that sync succeeded, failed or was cancelled has no bearing +/// on the caller's own sync, so a failure is only logged. pub(crate) async fn acquire_wallet_sync_guard<'a>( status: &'a Mutex, cancellation_error: Error, logger: &Logger, ) -> Result, Error> { @@ -125,11 +125,17 @@ pub(crate) async fn acquire_wallet_sync_guard<'a>( } }; log_debug!(logger, "Sync in progress, waiting for it to finish before syncing again."); - pending_sync.recv().await.map_err(|e| { - debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); - log_error!(logger, "Failed to receive wallet sync result: {:?}", e); - cancellation_error - })??; + match pending_sync.recv().await { + Ok(Ok(())) => {}, + Ok(Err(e)) => { + log_debug!(logger, "Sync in progress failed with {}, syncing again regardless.", e); + }, + Err(e) => { + debug_assert!(false, "Failed to receive wallet sync result: {:?}", e); + log_error!(logger, "Failed to receive wallet sync result: {:?}", e); + return Err(cancellation_error); + }, + } } } From 2cab32770add5499fb5a79c8bff0624ac2fada55 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 20 Oct 2026 17:44:34 -0500 Subject: [PATCH 08/10] Upload the LDK node's logs when an interop job fails The eclair job failed with `sync_wallets failed: WalletOperationFailed` on three unrelated branches on the same day. The node logs which part of the sync failed, but only to its log file under a random directory in the runner's temp dir, which the job discards. So each occurrence leaves nothing to tell a transient electrs error from a failed background sync the manual call piggybacked on. Point the harness's temp dir at a directory under the runner's temp and upload it as an artifact when the tests fail, together with the electrs and external node container logs for the same window. The same for the CLN and LND jobs, which run on the same harness. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/cln-integration.yml | 21 +++++++++++++++++++-- .github/workflows/eclair-integration.yml | 19 ++++++++++++++++++- .github/workflows/lnd-integration.yml | 21 +++++++++++++++++++-- 3 files changed, 56 insertions(+), 5 deletions(-) diff --git a/.github/workflows/cln-integration.yml b/.github/workflows/cln-integration.yml index daf8355bdd..6928385001 100644 --- a/.github/workflows/cln-integration.yml +++ b/.github/workflows/cln-integration.yml @@ -48,7 +48,24 @@ jobs: CLN_DATA_DIR: ${{ env.CLN_DATA_DIR }} - name: Run CLN integration tests - run: CLN_SOCKET_PATH=$CLN_DATA_DIR/regtest/lightning-rpc - RUSTFLAGS="--cfg cln_test" cargo test --test integration_tests_cln -- --show-output --test-threads=1 + run: | + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + TMPDIR="$RUNNER_TEMP/ldk-node-storage" CLN_SOCKET_PATH=$CLN_DATA_DIR/regtest/lightning-rpc RUSTFLAGS="--cfg cln_test" cargo test --test integration_tests_cln -- --show-output --test-threads=1 env: CLN_DATA_DIR: ${{ env.CLN_DATA_DIR }} + + - name: Collect service logs + if: failure() + run: | + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + docker compose -p ldk-node -f tests/docker/docker-compose-cln.yml logs --no-color --timestamps electrs > "$RUNNER_TEMP/ldk-node-storage/electrs.log" || true + docker compose -p ldk-node -f tests/docker/docker-compose-cln.yml logs --no-color --timestamps cln > "$RUNNER_TEMP/ldk-node-storage/cln.log" || true + + - name: Upload LDK node storage and service logs + if: failure() + uses: actions/upload-artifact@v7 + with: + name: cln-ldk-node-storage + path: ${{ runner.temp }}/ldk-node-storage + if-no-files-found: ignore + retention-days: 14 diff --git a/.github/workflows/eclair-integration.yml b/.github/workflows/eclair-integration.yml index 68acc8b13e..3c0db3ff49 100644 --- a/.github/workflows/eclair-integration.yml +++ b/.github/workflows/eclair-integration.yml @@ -67,4 +67,21 @@ jobs: - name: Run Eclair integration tests run: | - RUSTFLAGS="${{ matrix.rustflags }}" cargo test --test integration_tests_eclair -- --show-output --test-threads=1 + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + TMPDIR="$RUNNER_TEMP/ldk-node-storage" RUSTFLAGS="${{ matrix.rustflags }}" cargo test --test integration_tests_eclair -- --show-output --test-threads=1 + + - name: Collect service logs + if: failure() + run: | + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + docker compose -p ldk-node -f tests/docker/docker-compose-eclair.yml logs --no-color --timestamps electrs > "$RUNNER_TEMP/ldk-node-storage/electrs.log" || true + docker compose -p ldk-node -f tests/docker/docker-compose-eclair.yml logs --no-color --timestamps eclair > "$RUNNER_TEMP/ldk-node-storage/eclair.log" || true + + - name: Upload LDK node storage and service logs + if: failure() + uses: actions/upload-artifact@v7 + with: + name: eclair-${{ matrix.name }}-ldk-node-storage + path: ${{ runner.temp }}/ldk-node-storage + if-no-files-found: ignore + retention-days: 14 diff --git a/.github/workflows/lnd-integration.yml b/.github/workflows/lnd-integration.yml index 6c755e2a21..02037a443a 100644 --- a/.github/workflows/lnd-integration.yml +++ b/.github/workflows/lnd-integration.yml @@ -58,7 +58,24 @@ jobs: LND_DATA_DIR: ${{ env.LND_DATA_DIR }} - name: Run LND integration tests - run: LND_CERT_PATH=$LND_DATA_DIR/tls.cert LND_MACAROON_PATH=$LND_DATA_DIR/data/chain/bitcoin/regtest/admin.macaroon - RUSTFLAGS="--cfg lnd_test" cargo test --test integration_tests_lnd -- --show-output --test-threads=1 + run: | + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + TMPDIR="$RUNNER_TEMP/ldk-node-storage" LND_CERT_PATH=$LND_DATA_DIR/tls.cert LND_MACAROON_PATH=$LND_DATA_DIR/data/chain/bitcoin/regtest/admin.macaroon RUSTFLAGS="--cfg lnd_test" cargo test --test integration_tests_lnd -- --show-output --test-threads=1 env: LND_DATA_DIR: ${{ env.LND_DATA_DIR }} + + - name: Collect service logs + if: failure() + run: | + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + docker compose -p ldk-node -f tests/docker/docker-compose-lnd.yml logs --no-color --timestamps electrs > "$RUNNER_TEMP/ldk-node-storage/electrs.log" || true + docker compose -p ldk-node -f tests/docker/docker-compose-lnd.yml logs --no-color --timestamps lnd > "$RUNNER_TEMP/ldk-node-storage/lnd.log" || true + + - name: Upload LDK node storage and service logs + if: failure() + uses: actions/upload-artifact@v7 + with: + name: lnd-ldk-node-storage + path: ${{ runner.temp }}/ldk-node-storage + if-no-files-found: ignore + retention-days: 14 From f17b7c7e0e3d1875c93b3418f831c1a54add93bd Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 20 Oct 2026 19:22:24 -0500 Subject: [PATCH 09/10] Say what a poll was waiting for when it times out Every time-budgeted wait in the test harness panics from the same line with the same message, so a timeout in `reorg_test` on CI doesn't tell whether the test was waiting for electrs to serve a block, to see a transaction, or for a node's sweep balance to change. The waits that ran out on the macOS runners were all after the same block generation, and the message could not distinguish the candidates. Callers now say what they are waiting for, and the timeout message repeats it. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- tests/common/mod.rs | 49 +++++++++++++++++++++++++++++---------------- tests/reorg_test.rs | 2 +- 2 files changed, 33 insertions(+), 18 deletions(-) diff --git a/tests/common/mod.rs b/tests/common/mod.rs index bc6d726b3c..d44eefd467 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -933,21 +933,26 @@ pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) { pub(crate) async fn wait_for_block( bitcoind: &BitcoindClient, electrs: &E, min_height: usize, ) { - let expected_block_hash = exponential_backoff_poll(|| { - let bitcoind_height = - bitcoind.get_blockchain_info().expect("failed to get blockchain info").blocks as usize; - if bitcoind_height < min_height { - return None; - } - bitcoind.get_block_hash(min_height as u64).ok()?.block_hash().ok() - }) - .await; + let expected_block_hash = + exponential_backoff_poll(format!("bitcoind to reach height {}", min_height), || { + let bitcoind_height = + bitcoind.get_blockchain_info().expect("failed to get blockchain info").blocks + as usize; + if bitcoind_height < min_height { + return None; + } + bitcoind.get_block_hash(min_height as u64).ok()?.block_hash().ok() + }) + .await; // A height-only wait can return the old header during a same-height reorg. Require the // replacement hash so callers cannot sync against the stale chain by mistake. - exponential_backoff_poll(|| { - let header = electrs.block_header(min_height).ok()?; - (header.block_hash() == expected_block_hash).then_some(()) - }) + exponential_backoff_poll( + format!("electrs to serve block {} at height {}", expected_block_hash, min_height), + || { + let header = electrs.block_header(min_height).ok()?; + (header.block_hash() == expected_block_hash).then_some(()) + }, + ) .await; } @@ -956,7 +961,7 @@ pub(crate) async fn wait_for_tx(electrs: &E, txid: Txid) { return; } - exponential_backoff_poll(|| { + exponential_backoff_poll(format!("electrs to see transaction {}", txid), || { electrs.ping().unwrap(); electrs.transaction_get(&txid).ok() }) @@ -969,7 +974,7 @@ pub(crate) async fn wait_for_outpoint_spend(electrs: &E, outpoin // Script history already contains the funding transaction itself, so wait until the exact // funding outpoint leaves the unspent set instead of treating any history as a spend. - exponential_backoff_poll(|| { + exponential_backoff_poll(format!("electrs to see {} spent", outpoint), || { electrs.ping().unwrap(); let is_spent = !electrs.script_list_unspent(&txout_script).unwrap().iter().any(|output| { @@ -1011,7 +1016,12 @@ pub(crate) async fn wait_for_channel_ready_to_send( ); } -pub(crate) async fn exponential_backoff_poll(mut poll: F) -> T +/// Polls until `poll` returns a value. `waiting_for` says what the caller is waiting for, so a +/// timeout names it: the budget is shared by every wait in a test, and the panic line alone +/// doesn't tell which one ran out. +pub(crate) async fn exponential_backoff_poll( + waiting_for: impl std::fmt::Display, mut poll: F, +) -> T where F: FnMut() -> Option, { @@ -1029,7 +1039,12 @@ where None => {}, } - assert!(tokio::time::Instant::now() < deadline, "Timed out after {}s.", POLL_TIMEOUT_SECS); + assert!( + tokio::time::Instant::now() < deadline, + "Timed out after {}s waiting for {}.", + POLL_TIMEOUT_SECS, + waiting_for + ); tokio::time::sleep(delay).await; } } diff --git a/tests/reorg_test.rs b/tests/reorg_test.rs index aa17f072c3..0798b2e3a9 100644 --- a/tests/reorg_test.rs +++ b/tests/reorg_test.rs @@ -78,7 +78,7 @@ async fn wait_for_pending_sweep_balance( where F: FnMut(&PendingSweepBalance) -> bool, { - exponential_backoff_poll(|| { + exponential_backoff_poll(format!("a pending sweep balance on node {}", node.node_id()), || { node.sync_wallets().unwrap(); node.list_balances() .pending_balances_from_channel_closures From 2337e37bee2ce0daa934df84ae5faa826c404dd7 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Tue, 20 Oct 2026 19:22:26 -0500 Subject: [PATCH 10/10] Upload the test nodes' storage when a CI test job fails `reorg_test` times out on the macOS runners at the same point in most of the recent failing runs across unrelated PRs, after the maturity blocks are mined and while waiting for the first sweep. Locally that wait takes under five seconds, so the cause is only visible on the runner, and the node logs that would show it are discarded with the job. The test step now keeps the nodes' storage directories under the runner's temp dir and uploads them as an artifact on failure, as the interop jobs already do. Written with the help of an AI tool. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/rust.yml | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index a23120bf9a..8c46bbd5a3 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -87,11 +87,21 @@ jobs: - name: Test on Rust ${{ matrix.toolchain }} if: "matrix.platform != 'windows-latest'" run: | - RUSTFLAGS="--cfg no_download --cfg cycle_tests --cfg tokio_unstable" cargo test + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + TMPDIR="$RUNNER_TEMP/ldk-node-storage" RUSTFLAGS="--cfg no_download --cfg cycle_tests --cfg tokio_unstable" cargo test - name: Test with UniFFI support on Rust ${{ matrix.toolchain }} if: "matrix.platform != 'windows-latest' && matrix.build-uniffi" run: | - RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test --no-default-features --features uniffi-default + mkdir -p "$RUNNER_TEMP/ldk-node-storage" + TMPDIR="$RUNNER_TEMP/ldk-node-storage" RUSTFLAGS="--cfg no_download --cfg cycle_tests" cargo test --no-default-features --features uniffi-default + - name: Upload LDK node storage of the test nodes + if: "failure() && matrix.platform != 'windows-latest'" + uses: actions/upload-artifact@v7 + with: + name: ${{ matrix.platform }}-${{ matrix.toolchain }}-ldk-node-storage + path: ${{ runner.temp }}/ldk-node-storage + if-no-files-found: ignore + retention-days: 14 linting: name: Linting