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/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/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..0eddde1ae8 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,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 { + 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