From e18c6cde20930ebf984a350ba2b773a586098614 Mon Sep 17 00:00:00 2001 From: Jeffrey Czyz Date: Thu, 15 Oct 2026 13:45:23 -0500 Subject: [PATCH 1/6] 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 2/6] 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 3/6] 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 4/6] 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 5/6] 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 6/6] 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,