Make the test harness's chain waits tolerate slow CI runners

.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

.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

.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

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<Wallet>, channel_manager: Arc<ChannelManager>,
 		chain_monitor: Arc<ChainMonitor>, output_sweeper: Arc<Sweeper>,
 	) -> 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,54 @@ mod tests {
 		acquired_guard.complete(Ok(()));
 	}
 
+	/// 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<Result<(), Error>>,
+	) {
+		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");
+
+		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")
+			.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_waits_for_the_one_in_progress_and_then_runs_its_own() {
+		assert_a_sync_runs_after_the_one_in_progress_ends(Some(Ok(()))).await;
+	}
+
+	#[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;
+	}
+
+	#[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! {
 		fn arbitrary_witness()(
 			witness_elements in vec(vec(any::<u8>(), 0..100), 0..20)

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<Wallet>,
 	) -> 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<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> 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;

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<Wallet>,
 	) -> 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<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
 		output_sweeper: Arc<Sweeper>,
 	) -> 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;

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,38 @@ 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. 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<WalletSyncStatus>, cancellation_error: Error, logger: &Logger,
+) -> Result<WalletSyncGuard<'a>, 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.");
+		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);
+			},
+		}
+	}
+}
+
 impl WalletSyncStatus {
 	fn register_or_subscribe_pending_sync(
 		&mut self,