Merge pull request #956 from Camillarhi/multi-lsp-support-follow-up

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

Author: tnull

src/config.rs

@@ -129,6 +129,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/event.rs

@@ -1375,14 +1375,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

@@ -191,6 +191,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;
@@ -755,33 +756,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;
 				});
 			}
 
@@ -811,6 +786,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(())
@@ -2485,6 +2493,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,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