Follow-up to #792 (multi-LSP support)

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/event.rs

@@ -1327,14 +1327,15 @@ where
 				);
 				let mut allow_0conf =
 					self.config.trusted_peers_0conf.contains(&counterparty_node_id);
-				let mut channel_override_config = None;
 
 				// If the peer is a configured LSP node, additionally honor its trust_peer_0conf flag.
-				if let Some(lsp) =
-					self.liquidity_source.get_lsp_config(&counterparty_node_id, 2).await
-				{
-					allow_0conf = allow_0conf || lsp.trust_peer_0conf;
+				if self.liquidity_source.get_lsp_trust_0conf(&counterparty_node_id) == Some(true) {
+					allow_0conf = true;
+				}
+
+				let mut channel_override_config = None;
 
+				if self.liquidity_source.get_lsp_config(&counterparty_node_id, 2).await.is_some() {
 					// When we're an LSPS2 client, allow claiming underpaying HTLCs as the LSP will skim off some fee. We'll
 					// check that they don't take too much before claiming.
 					channel_override_config = Some(ChannelConfigOverrides {

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

@@ -517,6 +517,25 @@ 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()
+	}
+
+	pub(crate) fn get_lsp_trust_0conf(&self, node_id: &PublicKey) -> Option<bool> {
+		self.lsp_nodes
+			.read()
+			.expect("lock")
+			.iter()
+			.find(|n| &n.node_id == node_id)
+			.map(|n| n.trust_peer_0conf)
+	}
+
 	/// Flips the `discovery_done` watch to `true`.
 	///
 	/// Called once after the *initial* batch of LSPs configured at build time has been