From a201d401be18b107b219592f8281aed69331e19d Mon Sep 17 00:00:00 2001 From: Camillarhi Date: Wed, 1 Jul 2026 11:20:21 +0100 Subject: [PATCH 1/2] 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. --- src/config.rs | 7 ++++ src/lib.rs | 79 +++++++++++++++++++++++++++++--------------- src/liquidity/mod.rs | 52 ++++++++++++++++------------- 3 files changed, 88 insertions(+), 50 deletions(-) diff --git a/src/config.rs b/src/config.rs index f168df94ef..fde04cee27 100644 --- a/src/config.rs +++ b/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. diff --git a/src/lib.rs b/src/lib.rs index acfcbc0d48..7a6dbac1e2 100644 --- a/src/lib.rs +++ b/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>, + liquidity_source: &LiquiditySource>, 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}; diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index 87a0650c83..9079de40f3 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -345,7 +345,7 @@ where lsps1_client: Arc>, lsps2_client: Arc>, lsps2_service: Arc>, - pending_lsps0_discovery: Mutex>>>, + pending_lsps0_discovery: Mutex>>>>, discovery_done_tx: tokio::sync::watch::Sender, discovery_done_rx: tokio::sync::watch::Receiver, liquidity_manager: Arc, @@ -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 From 48104f78aba1d555653bae0208b8bccc232024ad Mon Sep 17 00:00:00 2001 From: Camillarhi Date: Wed, 1 Jul 2026 12:03:24 +0100 Subject: [PATCH 2/2] Honor LSP trust_peer_0conf independent of supported protocols Look up trust_peer_0conf by node id via a protocol-independent helper that does not depend on discovery --- src/event.rs | 11 ++++++----- src/liquidity/mod.rs | 9 +++++++++ 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/src/event.rs b/src/event.rs index 91ab7b27de..9be465c22a 100644 --- a/src/event.rs +++ b/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 { diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index 9079de40f3..0eddde1ae8 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -523,6 +523,15 @@ where .collect() } + pub(crate) fn get_lsp_trust_0conf(&self, node_id: &PublicKey) -> Option { + 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