Retry LSP protocol discovery and periodically re-discover

Add a background task that retries discovery for any LSP whose protocols
are still undiscovered, using exponential backoff (5s up to 1h). Once no
undiscovered LSPs remain (or the backoff is exhausted), the task settles
into a fixed interval (24h) and re-runs discovery for all configured
LSPs, so we also pick up protocols an LSP rolls out after we first
connected.

Author: Camillarhi

src/config.rs

@@ -128,6 +128,17 @@ pub(crate) const HRN_RESOLUTION_TIMEOUT_SECS: u64 = 5;
 // The timeout after which we abort an LNURL-auth operation.
 pub(crate) const LNURL_AUTH_TIMEOUT_SECS: u64 = 15;
 
+// The initial delay before retrying a failed liquidity protocol discovery operation.
+pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::from_secs(5);
+
+// The maximum delay the initial discovery-retry backoff ramps up to before handing off to the
+// periodic re-discovery sweep.
+pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60);
+
+// The interval at which we re-run protocol discovery for our configured LSPs, to pick up
+// protocols an LSP may have rolled out since the last discovery.
+pub(crate) const LIQUIDITY_REDISCOVERY_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
+
 #[derive(Debug, Clone)]
 #[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
 /// Represents the configuration of an [`Node`] instance.

src/lib.rs

@@ -189,6 +189,10 @@ pub use types::{
 };
 pub use vss_client;
 
+use crate::config::{
+	LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY,
+	LIQUIDITY_REDISCOVERY_INTERVAL,
+};
 use crate::ffi::maybe_wrap;
 use crate::liquidity::Liquidity;
 use crate::scoring::setup_background_pathfinding_scores_sync;
@@ -706,33 +710,7 @@ impl Node {
 				let logger = Arc::clone(&discovery_logger);
 				let ls = Arc::clone(&liquidity_handler);
 				discovery_set.spawn(async move {
-					if let Err(e) = cm.connect_peer_if_necessary(node_id, address.clone()).await {
-						log_error!(
-							logger,
-							"Failed to connect to LSP {} for protocol discovery: {}",
-							node_id,
-							e
-						);
-						return;
-					}
-					match ls.discover_lsp_protocols(&node_id).await {
-						Ok(protocols) => {
-							log_info!(
-								logger,
-								"Discovered protocols for LSP {}: {:?}",
-								node_id,
-								protocols
-							);
-						},
-						Err(e) => {
-							log_error!(
-								logger,
-								"Failed to discover protocols for LSP {}: {:?}",
-								node_id,
-								e
-							);
-						},
-					}
+					connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
 				});
 			}
 
@@ -762,6 +740,65 @@ impl Node {
 			}
 		});
 
+		// Retry protocol discovery for any LSPs that failed the startup batch, backing off
+		// until we reach the periodic re-discovery cadence. From then on, re-discover all
+		// configured LSPs on a fixed interval to pick up protocols they roll out later.
+		let mut stop_rediscovery = self.stop_sender.subscribe();
+		let rediscovery_ls = Arc::clone(&self.liquidity_source);
+		let rediscovery_logger = Arc::clone(&self.logger);
+		let rediscovery_cm = Arc::clone(&self.connection_manager);
+		self.runtime.spawn_cancellable_background_task(async move {
+			// Fast retries for LSPs that failed the startup discovery batch, backing off
+			// until we reach the periodic re-discovery cadence.
+			let mut backoff = LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY;
+			while backoff < LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY {
+				tokio::select! {
+						_ = stop_rediscovery.changed() => return,
+						_ = tokio::time::sleep(backoff) => {},
+				}
+
+				let undiscovered_lsps = rediscovery_ls.get_undiscovered_lsps();
+				if undiscovered_lsps.is_empty() {
+					break;
+				}
+
+				let mut discovery_set = tokio::task::JoinSet::new();
+				for (node_id, address) in undiscovered_lsps {
+					let cm = Arc::clone(&rediscovery_cm);
+					let ls = Arc::clone(&rediscovery_ls);
+					let logger = Arc::clone(&rediscovery_logger);
+					discovery_set.spawn(async move {
+						connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
+					});
+				}
+				discovery_set.join_all().await;
+				backoff *= 2;
+			}
+
+			// periodically re-discover all configured LSPs to pick up newly
+			// rolled-out protocols and recover any nodes that never completed discovery.
+			let mut interval = tokio::time::interval(LIQUIDITY_REDISCOVERY_INTERVAL);
+			interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+			interval.tick().await;
+			loop {
+				tokio::select! {
+						_ = stop_rediscovery.changed() => return,
+						_ = interval.tick() => {},
+				}
+
+				let mut discovery_set = tokio::task::JoinSet::new();
+				for (node_id, address) in rediscovery_ls.get_all_lsp_details() {
+					let cm = Arc::clone(&rediscovery_cm);
+					let ls = Arc::clone(&rediscovery_ls);
+					let logger = Arc::clone(&rediscovery_logger);
+					discovery_set.spawn(async move {
+						connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
+					});
+				}
+				discovery_set.join_all().await;
+			}
+		});
+
 		log_info!(self.logger, "Startup complete.");
 		*is_running_lock = true;
 		Ok(())
@@ -2435,6 +2472,23 @@ pub(crate) fn new_channel_anchor_reserve_sats(
 	})
 }
 
+async fn connect_and_discover_lsp(
+	connection_manager: &ConnectionManager<Arc<Logger>>,
+	liquidity_source: &LiquiditySource<Arc<Logger>>, logger: &Logger, node_id: PublicKey,
+	address: SocketAddress,
+) {
+	if let Err(e) = connection_manager.connect_peer_if_necessary(node_id, address).await {
+		log_debug!(logger, "Failed to connect to LSP {} for protocol discovery: {}", node_id, e);
+		return;
+	}
+	match liquidity_source.discover_lsp_protocols(&node_id).await {
+		Ok(protocols) => {
+			log_info!(logger, "Discovered protocols for LSP {}: {:?}", node_id, protocols)
+		},
+		Err(e) => log_debug!(logger, "Protocol discovery failed for LSP {}: {:?}", node_id, e),
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use lightning::util::ser::{Readable, Writeable};

src/liquidity/mod.rs

@@ -352,6 +352,21 @@ where
 	logger: L,
 }
 
+// Removes the pending LSPS0 discovery entry for a node when dropped, ensuring cleanup happens
+// even if the discovery future is cancelled mid-await (e.g., when `Node::stop` aborts the
+// background task). Otherwise a stale entry would reject any subsequent discovery attempt for
+// this node as already being in flight.
+struct PendingDiscoveryGuard<'a> {
+	pending: &'a Mutex<HashMap<PublicKey, oneshot::Sender<Vec<u16>>>>,
+	node_id: PublicKey,
+}
+
+impl Drop for PendingDiscoveryGuard<'_> {
+	fn drop(&mut self) {
+		self.pending.lock().expect("lock").remove(&self.node_id);
+	}
+}
+
 impl<L: Deref> LiquiditySource<L>
 where
 	L::Target: LdkLogger,
@@ -456,6 +471,10 @@ where
 			}
 		}
 
+		// Clean up the pending entry on every exit path, including cancellation of this future.
+		let _guard =
+			PendingDiscoveryGuard { pending: &self.pending_lsps0_discovery, node_id: *node_id };
+
 		let protocols =
 			tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver)
 				.await
@@ -466,7 +485,6 @@ where
 						node_id,
 						e
 					);
-					self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
 					Error::LiquidityRequestFailed
 				})?
 				.map_err(|e| {
@@ -476,7 +494,6 @@ where
 						node_id,
 						e
 					);
-					self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
 					Error::LiquidityRequestFailed
 				})?;
 
@@ -517,6 +534,16 @@ where
 		select_lsps_for_protocol(&self.lsp_nodes, protocol, Some(node_id))
 	}
 
+	pub(crate) fn get_undiscovered_lsps(&self) -> Vec<(PublicKey, SocketAddress)> {
+		self.lsp_nodes
+			.read()
+			.expect("lock")
+			.iter()
+			.filter(|n| n.supported_protocols.is_none())
+			.map(|n| (n.node_id, n.address.clone()))
+			.collect()
+	}
+
 	/// Flips the `discovery_done` watch to `true`.
 	///
 	/// Called once after the *initial* batch of LSPs configured at build time has been