Retry failed LSP protocol discovery

Add a background task that retries discovery for LSPs still undiscovered
after the startup batch, with exponential backoff (5s up to a 1h cap),
until every configured LSP is discovered. This recovers LSPs that were
unreachable at startup instead of leaving them permanently unusable.

Also coalesce concurrent discovery for the same LSP, so the retry task
and a runtime add_liquidity_source don't race and needlessly fail.

Author: Camillarhi

src/config.rs

@@ -128,6 +128,13 @@ 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 discovery-retry backoff ramps up to, and the interval it keeps retrying at
+// thereafter until every configured LSP has been discovered.
+pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(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,7 @@ pub use types::{
 };
 pub use vss_client;
 
+use crate::config::{LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY};
 use crate::ffi::maybe_wrap;
 use crate::liquidity::Liquidity;
 use crate::scoring::setup_background_pathfinding_scores_sync;
@@ -706,33 +707,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 +737,39 @@ impl Node {
 			}
 		});
 
+		// Retry protocol discovery for any LSPs that failed the startup batch, backing off up to a
+		// cap and then retrying at that cap until every configured LSP has been discovered.
+		let mut stop_retry = self.stop_sender.subscribe();
+		let retry_ls = Arc::clone(&self.liquidity_source);
+		let retry_logger = Arc::clone(&self.logger);
+		let retry_cm = Arc::clone(&self.connection_manager);
+		self.runtime.spawn_cancellable_background_task(async move {
+			let mut backoff = LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY;
+			loop {
+				tokio::select! {
+						_ = stop_retry.changed() => return,
+						_ = tokio::time::sleep(backoff) => {},
+				}
+
+				let undiscovered_lsps = retry_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(&retry_cm);
+					let ls = Arc::clone(&retry_ls);
+					let logger = Arc::clone(&retry_logger);
+					discovery_set.spawn(async move {
+						connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
+					});
+				}
+				discovery_set.join_all().await;
+				backoff = (backoff * 2).min(LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY);
+			}
+		});
+
 		log_info!(self.logger, "Startup complete.");
 		*is_running_lock = true;
 		Ok(())
@@ -2435,6 +2443,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

@@ -345,7 +345,7 @@ where
 	lsps1_client: Arc<LSPS1Client<L>>,
 	lsps2_client: Arc<LSPS2Client<L>>,
 	lsps2_service: Arc<LSPS2ServiceLiquiditySource<L>>,
-	pending_lsps0_discovery: Mutex<HashMap<PublicKey, oneshot::Sender<Vec<u16>>>>,
+	pending_lsps0_discovery: Mutex<HashMap<PublicKey, Vec<oneshot::Sender<Vec<u16>>>>>,
 	discovery_done_tx: tokio::sync::watch::Sender<bool>,
 	discovery_done_rx: tokio::sync::watch::Receiver<bool>,
 	liquidity_manager: Arc<LiquidityManager>,
@@ -383,21 +383,14 @@ where
 				protocols,
 			}) => {
 				if self.is_lsps_node(&counterparty_node_id) {
-					if let Some(sender) = self
+					if let Some(senders) = self
 						.pending_lsps0_discovery
 						.lock()
 						.expect("lock")
 						.remove(&counterparty_node_id)
 					{
-						match sender.send(protocols) {
-							Ok(()) => (),
-							Err(_) => {
-								log_error!(
-									self.logger,
-									"Failed to handle response for request {:?} from liquidity service",
-									counterparty_node_id
-								);
-							},
+						for sender in senders {
+							let _ = sender.send(protocols.clone());
 						}
 					} else {
 						log_error!(
@@ -438,24 +431,23 @@ where
 		let lsps0_handler = self.liquidity_manager.lsps0_client_handler();
 
 		let (sender, receiver) = oneshot::channel();
-		{
+		let issued_request = {
 			let mut pending_discovery = self.pending_lsps0_discovery.lock().expect("lock");
 			match pending_discovery.entry(*node_id) {
-				Entry::Occupied(_) => {
-					log_error!(
-						self.logger,
-						"LSPS0 protocol discovery already in flight for {}",
-						node_id
-					);
-					return Err(Error::LiquidityRequestFailed);
+				Entry::Occupied(mut e) => {
+					e.get_mut().push(sender);
+					false
 				},
 				Entry::Vacant(v) => {
-					v.insert(sender);
+					v.insert(vec![sender]);
 					lsps0_handler.list_protocols(node_id);
+					true
 				},
 			}
-		}
+		};
 
+		// Only the request that issued the discovery may remove the entry; a follower removing it
+		// would drop the in-flight request and all other waiters.
 		let protocols =
 			tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver)
 				.await
@@ -466,7 +458,9 @@ where
 						node_id,
 						e
 					);
-					self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
+					if issued_request {
+						self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
+					}
 					Error::LiquidityRequestFailed
 				})?
 				.map_err(|e| {
@@ -476,7 +470,9 @@ where
 						node_id,
 						e
 					);
-					self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
+					if issued_request {
+						self.pending_lsps0_discovery.lock().expect("lock").remove(node_id);
+					}
 					Error::LiquidityRequestFailed
 				})?;
 
@@ -517,6 +513,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