Release completed cancellable tasks

Cargo.toml

@@ -70,6 +70,7 @@ base64 = { version = "0.22.1", default-features = false, features = ["std"] }
 getrandom = { version = "0.3", default-features = false }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 tokio = { version = "1.39", default-features = false, features = [ "rt-multi-thread", "time", "sync", "macros", "net" ] }
+tokio-util = { version = "0.7", default-features = false, features = ["rt"] }
 esplora-client = { version = "0.12", default-features = false, features = ["tokio", "async-https-rustls"] }
 electrum-client = { version = "0.25", default-features = false, features = ["proxy", "use-rustls-ring"] }
 libc = "0.2"

src/chain/bitcoind.rs

@@ -31,7 +31,7 @@ use lightning_block_sync::{
 };
 use serde::Serialize;
 
-use super::WalletSyncStatus;
+use super::{WalletSyncGuard, WalletSyncStatus};
 use crate::config::{
 	BitcoindRestClientConfig, Config, DEFAULT_FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS,
 	DEFAULT_TX_BROADCAST_TIMEOUT_SECS,
@@ -152,12 +152,14 @@ impl BitcoindChainSource {
 	) {
 		// First register for the wallet polling status to make sure `Node::sync_wallets` calls
 		// wait on the result before proceeding.
-		{
+		let initial_sync_guard = {
 			let mut status_lock = self.wallet_polling_status.lock().expect("lock");
 			if status_lock.register_or_subscribe_pending_sync().is_some() {
 				debug_assert!(false, "Sync already in progress. This should never happen.");
+				return;
 			}
-		}
+			WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed)
+		};
 
 		log_info!(
 			self.logger,
@@ -285,7 +287,7 @@ impl BitcoindChainSource {
 		}
 
 		// Now propagate the initial result to unblock waiting subscribers.
-		self.wallet_polling_status.lock().expect("lock").propagate_result_to_subscribers(Ok(()));
+		initial_sync_guard.complete(Ok(()));
 
 		let mut chain_polling_interval =
 			tokio::time::interval(Duration::from_secs(CHAIN_POLLING_INTERVAL_SECS));
@@ -396,6 +398,8 @@ impl BitcoindChainSource {
 				Error::WalletOperationFailed
 			})?;
 		}
+		let sync_guard =
+			WalletSyncGuard::new(&self.wallet_polling_status, Error::WalletOperationFailed);
 
 		let res = self
 			.poll_and_update_listeners_inner(
@@ -406,7 +410,7 @@ impl BitcoindChainSource {
 			)
 			.await;
 
-		self.wallet_polling_status.lock().expect("lock").propagate_result_to_subscribers(res);
+		sync_guard.complete(res);
 
 		res
 	}

src/chain/electrum.rs

@@ -25,7 +25,7 @@ use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
 use lightning_transaction_sync::ElectrumSyncClient;
 
-use super::WalletSyncStatus;
+use super::{WalletSyncGuard, WalletSyncStatus};
 use crate::config::{
 	clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, MAX_FULL_SCAN_STOP_GAP,
 	MIN_FULL_SCAN_STOP_GAP,
@@ -113,10 +113,12 @@ impl ElectrumChainSource {
 				Error::WalletOperationFailed
 			})?;
 		}
+		let sync_guard =
+			WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed);
 
 		let res = self.sync_onchain_wallet_inner(onchain_wallet).await;
 
-		self.onchain_wallet_sync_status.lock().expect("lock").propagate_result_to_subscribers(res);
+		sync_guard.complete(res);
 
 		res
 	}
@@ -223,14 +225,13 @@ impl ElectrumChainSource {
 				Error::TxSyncFailed
 			})?;
 		}
+		let sync_guard =
+			WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::TxSyncFailed);
 
 		let res =
 			self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await;
 
-		self.lightning_wallet_sync_status
-			.lock()
-			.expect("lock")
-			.propagate_result_to_subscribers(res);
+		sync_guard.complete(res);
 
 		res
 	}

src/chain/esplora.rs

@@ -18,7 +18,7 @@ use lightning::chain::{Confirm, Filter, WatchedOutput};
 use lightning::util::ser::Writeable;
 use lightning_transaction_sync::EsploraSyncClient;
 
-use super::WalletSyncStatus;
+use super::{WalletSyncGuard, WalletSyncStatus};
 use crate::config::{
 	clamp_full_scan_stop_gap, Config, EsploraSyncConfig, BDK_CLIENT_CONCURRENCY,
 	MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP,
@@ -125,10 +125,12 @@ impl EsploraChainSource {
 				Error::WalletOperationFailed
 			})?;
 		}
+		let sync_guard =
+			WalletSyncGuard::new(&self.onchain_wallet_sync_status, Error::WalletOperationFailed);
 
 		let res = self.sync_onchain_wallet_inner(onchain_wallet).await;
 
-		self.onchain_wallet_sync_status.lock().expect("lock").propagate_result_to_subscribers(res);
+		sync_guard.complete(res);
 
 		res
 	}
@@ -275,14 +277,13 @@ impl EsploraChainSource {
 				Error::WalletOperationFailed
 			})?;
 		}
+		let sync_guard =
+			WalletSyncGuard::new(&self.lightning_wallet_sync_status, Error::WalletOperationFailed);
 
 		let res =
 			self.sync_lightning_wallet_inner(channel_manager, chain_monitor, output_sweeper).await;
 
-		self.lightning_wallet_sync_status
-			.lock()
-			.expect("lock")
-			.propagate_result_to_subscribers(res);
+		sync_guard.complete(res);
 
 		res
 	}

src/chain/mod.rs

@@ -65,6 +65,34 @@ pub(crate) enum WalletSyncStatus {
 	InProgress { subscribers: tokio::sync::broadcast::Sender<Result<(), Error>> },
 }
 
+pub(crate) struct WalletSyncGuard<'a> {
+	status: &'a Mutex<WalletSyncStatus>,
+	cancellation_error: Error,
+	active: bool,
+}
+
+impl<'a> WalletSyncGuard<'a> {
+	pub(crate) fn new(status: &'a Mutex<WalletSyncStatus>, cancellation_error: Error) -> Self {
+		Self { status, cancellation_error, active: true }
+	}
+
+	pub(crate) fn complete(mut self, res: Result<(), Error>) {
+		self.status.lock().expect("lock").propagate_result_to_subscribers(res);
+		self.active = false;
+	}
+}
+
+impl Drop for WalletSyncGuard<'_> {
+	fn drop(&mut self) {
+		if self.active {
+			self.status
+				.lock()
+				.expect("lock")
+				.propagate_result_to_subscribers(Err(self.cancellation_error));
+		}
+	}
+}
+
 impl WalletSyncStatus {
 	fn register_or_subscribe_pending_sync(
 		&mut self,
@@ -95,16 +123,7 @@ impl WalletSyncStatus {
 				WalletSyncStatus::InProgress { subscribers } => {
 					// A sync is in-progress, we notify subscribers.
 					if subscribers.receiver_count() > 0 {
-						match subscribers.send(res) {
-							Ok(_) => (),
-							Err(e) => {
-								debug_assert!(
-									false,
-									"Failed to send wallet sync result to subscribers: {:?}",
-									e
-								);
-							},
-						}
+						let _ = subscribers.send(res);
 					}
 					*self = WalletSyncStatus::Completed;
 				},
@@ -561,3 +580,28 @@ impl Filter for ChainSource {
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn wallet_sync_guard_resets_abandoned_sync() {
+		let status = Mutex::new(WalletSyncStatus::Completed);
+		assert!(status.lock().expect("lock").register_or_subscribe_pending_sync().is_none());
+		let sync_guard = WalletSyncGuard::new(&status, Error::WalletOperationFailed);
+		let mut subscriber = status
+			.lock()
+			.expect("lock")
+			.register_or_subscribe_pending_sync()
+			.expect("sync subscriber");
+
+		drop(sync_guard);
+
+		assert!(
+			matches!(*status.lock().expect("lock"), WalletSyncStatus::Completed),
+			"abandoned wallet sync should reset its status"
+		);
+		assert_eq!(subscriber.try_recv(), Ok(Err(Error::WalletOperationFailed)));
+	}
+}

src/connection.rs

@@ -18,12 +18,44 @@ use crate::logger::{log_debug, log_error, log_info, LdkLogger};
 use crate::types::{KeysManager, PeerManager};
 use crate::Error;
 
+type PendingConnections =
+	Mutex<HashMap<PublicKey, Vec<tokio::sync::oneshot::Sender<Result<(), Error>>>>>;
+
+struct PendingConnectionGuard<'a> {
+	pending_connections: &'a PendingConnections,
+	node_id: PublicKey,
+	active: bool,
+}
+
+impl<'a> PendingConnectionGuard<'a> {
+	fn new(pending_connections: &'a PendingConnections, node_id: PublicKey) -> Self {
+		Self { pending_connections, node_id, active: true }
+	}
+
+	fn disarm(&mut self) {
+		self.active = false;
+	}
+}
+
+impl Drop for PendingConnectionGuard<'_> {
+	fn drop(&mut self) {
+		if !self.active {
+			return;
+		}
+		let mut pending_connections = self.pending_connections.lock().expect("lock");
+		if let Some(subscribers) = pending_connections.remove(&self.node_id) {
+			for subscriber in subscribers {
+				let _ = subscriber.send(Err(Error::ConnectionFailed));
+			}
+		}
+	}
+}
+
 pub(crate) struct ConnectionManager<L: Deref + Clone + Sync + Send>
 where
 	L::Target: LdkLogger,
 {
-	pending_connections:
-		Mutex<HashMap<PublicKey, Vec<tokio::sync::oneshot::Sender<Result<(), Error>>>>>,
+	pending_connections: PendingConnections,
 	peer_manager: Arc<PeerManager>,
 	tor_proxy_config: Option<TorConfig>,
 	keys_manager: Arc<KeysManager>,
@@ -60,26 +92,29 @@ where
 	pub(crate) async fn do_connect_peer(
 		&self, node_id: PublicKey, addr: SocketAddress,
 	) -> Result<(), Error> {
+		// If another task is already connecting, subscribe to its result instead of starting a
+		// duplicate attempt.
+		if let Some(pending_connection_ready_receiver) =
+			self.register_or_subscribe_pending_connection(&node_id)
+		{
+			return pending_connection_ready_receiver.await.map_err(|e| {
+				debug_assert!(false, "Failed to receive connection result: {:?}", e);
+				log_error!(self.logger, "Failed to receive connection result: {:?}", e);
+				Error::ConnectionFailed
+			})?;
+		}
+
+		let mut pending_connection =
+			PendingConnectionGuard::new(&self.pending_connections, node_id);
 		let res = self.do_connect_peer_internal(node_id, addr).await;
 		self.propagate_result_to_subscribers(&node_id, res);
+		pending_connection.disarm();
 		res
 	}
 
 	async fn do_connect_peer_internal(
 		&self, node_id: PublicKey, addr: SocketAddress,
 	) -> Result<(), Error> {
-		// First, we check if there is already an outbound connection in flight, if so, we just
-		// await on the corresponding watch channel. The task driving the connection future will
-		// send us the result..
-		let pending_ready_receiver_opt = self.register_or_subscribe_pending_connection(&node_id);
-		if let Some(pending_connection_ready_receiver) = pending_ready_receiver_opt {
-			return pending_connection_ready_receiver.await.map_err(|e| {
-				debug_assert!(false, "Failed to receive connection result: {:?}", e);
-				log_error!(self.logger, "Failed to receive connection result: {:?}", e);
-				Error::ConnectionFailed
-			})?;
-		}
-
 		log_info!(self.logger, "Connecting to peer: {}@{}", node_id, addr);
 
 		match addr {
@@ -246,6 +281,7 @@ where
 		match pending_connections_lock.entry(*node_id) {
 			hash_map::Entry::Occupied(mut entry) => {
 				let (tx, rx) = tokio::sync::oneshot::channel();
+				entry.get_mut().retain(|subscriber| !subscriber.is_closed());
 				entry.get_mut().push(tx);
 				Some(rx)
 			},
@@ -277,3 +313,26 @@ where
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn pending_connection_guard_notifies_subscribers_when_abandoned() {
+		let node_id: PublicKey =
+			"0279be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798".parse().unwrap();
+		let pending_connections = Mutex::new(HashMap::new());
+		let (sender, mut receiver) = tokio::sync::oneshot::channel();
+		pending_connections.lock().expect("lock").insert(node_id, vec![sender]);
+
+		let connection_guard = PendingConnectionGuard::new(&pending_connections, node_id);
+		drop(connection_guard);
+
+		assert!(
+			pending_connections.lock().expect("lock").is_empty(),
+			"abandoned connection attempt should remove pending state"
+		);
+		assert_eq!(receiver.try_recv(), Ok(Err(Error::ConnectionFailed)));
+	}
+}