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, diff --git a/tests/common/mod.rs b/tests/common/mod.rs index bb1f366b1c..bc6d726b3c 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; } } @@ -2027,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")] @@ -2051,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, @@ -2172,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; @@ -2224,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; @@ -2261,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")] 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); 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" 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);