Skip to content

Commit 70b5832

Browse files
committed
Retry LSP protocol discovery and periodically re-discover
Add a background task that retries discovery for any LSP whose protocols are still undiscovered, using exponential backoff (5s up to 1h). Once no undiscovered LSPs remain (or the backoff is exhausted), the task settles into a fixed interval (24h) and re-runs discovery for all configured LSPs, so we also pick up protocols an LSP rolls out after we first connected.
1 parent dc4fa49 commit 70b5832

3 files changed

Lines changed: 101 additions & 0 deletions

File tree

src/config.rs

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,17 @@ pub(crate) const HRN_RESOLUTION_TIMEOUT_SECS: u64 = 5;
128128
// The timeout after which we abort an LNURL-auth operation.
129129
pub(crate) const LNURL_AUTH_TIMEOUT_SECS: u64 = 15;
130130

131+
// The initial delay before retrying a failed liquidity protocol discovery operation.
132+
pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::from_secs(5);
133+
134+
// The maximum delay the initial discovery-retry backoff ramps up to before handing off to the
135+
// periodic re-discovery sweep.
136+
pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60);
137+
138+
// The interval at which we re-run protocol discovery for our configured LSPs, to pick up
139+
// protocols an LSP may have rolled out since the last discovery.
140+
pub(crate) const LIQUIDITY_REDISCOVERY_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
141+
131142
#[derive(Debug, Clone)]
132143
#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
133144
/// Represents the configuration of an [`Node`] instance.

src/lib.rs

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,10 @@ pub use types::{
189189
};
190190
pub use vss_client;
191191

192+
use crate::config::{
193+
LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY,
194+
LIQUIDITY_REDISCOVERY_INTERVAL,
195+
};
192196
use crate::ffi::maybe_wrap;
193197
use crate::liquidity::Liquidity;
194198
use crate::scoring::setup_background_pathfinding_scores_sync;
@@ -762,6 +766,65 @@ impl Node {
762766
}
763767
});
764768

769+
// Retry protocol discovery for any LSPs that failed the startup batch, backing off
770+
// until we reach the periodic re-discovery cadence. From then on, re-discover all
771+
// configured LSPs on a fixed interval to pick up protocols they roll out later.
772+
let mut stop_rediscovery = self.stop_sender.subscribe();
773+
let rediscovery_ls = Arc::clone(&self.liquidity_source);
774+
let rediscovery_logger = Arc::clone(&self.logger);
775+
let rediscovery_cm = Arc::clone(&self.connection_manager);
776+
self.runtime.spawn_cancellable_background_task(async move {
777+
// Fast retries for LSPs that failed the startup discovery batch, backing off
778+
// until we reach the periodic re-discovery cadence.
779+
let mut backoff = LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY;
780+
while backoff < LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY {
781+
tokio::select! {
782+
_ = stop_rediscovery.changed() => return,
783+
_ = tokio::time::sleep(backoff) => {},
784+
}
785+
786+
let undiscovered_lsps = rediscovery_ls.get_undiscovered_lsps();
787+
if undiscovered_lsps.is_empty() {
788+
break;
789+
}
790+
791+
let mut discovery_set = tokio::task::JoinSet::new();
792+
for (node_id, address) in undiscovered_lsps {
793+
let cm = Arc::clone(&rediscovery_cm);
794+
let ls = Arc::clone(&rediscovery_ls);
795+
let logger = Arc::clone(&rediscovery_logger);
796+
discovery_set.spawn(async move {
797+
connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
798+
});
799+
}
800+
discovery_set.join_all().await;
801+
backoff *= 2;
802+
}
803+
804+
// periodically re-discover all configured LSPs to pick up newly
805+
// rolled-out protocols and recover any nodes that never completed discovery.
806+
let mut interval = tokio::time::interval(LIQUIDITY_REDISCOVERY_INTERVAL);
807+
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
808+
interval.tick().await;
809+
loop {
810+
tokio::select! {
811+
_ = stop_rediscovery.changed() => return,
812+
_ = interval.tick() => {},
813+
}
814+
815+
let mut discovery_set = tokio::task::JoinSet::new();
816+
for (node_id, address) in rediscovery_ls.get_all_lsp_details() {
817+
let cm = Arc::clone(&rediscovery_cm);
818+
let ls = Arc::clone(&rediscovery_ls);
819+
let logger = Arc::clone(&rediscovery_logger);
820+
discovery_set.spawn(async move {
821+
connect_and_discover_lsp(&cm, &ls, &logger, node_id, address).await;
822+
});
823+
}
824+
discovery_set.join_all().await;
825+
}
826+
});
827+
765828
log_info!(self.logger, "Startup complete.");
766829
*is_running_lock = true;
767830
Ok(())
@@ -2435,6 +2498,23 @@ pub(crate) fn new_channel_anchor_reserve_sats(
24352498
})
24362499
}
24372500

2501+
async fn connect_and_discover_lsp(
2502+
connection_manager: &ConnectionManager<Arc<Logger>>,
2503+
liquidity_source: &LiquiditySource<Arc<Logger>>, logger: &Logger, node_id: PublicKey,
2504+
address: SocketAddress,
2505+
) {
2506+
if let Err(e) = connection_manager.connect_peer_if_necessary(node_id, address).await {
2507+
log_debug!(logger, "Failed to connect to LSP {} for protocol discovery: {}", node_id, e);
2508+
return;
2509+
}
2510+
match liquidity_source.discover_lsp_protocols(&node_id).await {
2511+
Ok(protocols) => {
2512+
log_info!(logger, "Discovered protocols for LSP {}: {:?}", node_id, protocols)
2513+
},
2514+
Err(e) => log_debug!(logger, "Protocol discovery failed for LSP {}: {:?}", node_id, e),
2515+
}
2516+
}
2517+
24382518
#[cfg(test)]
24392519
mod tests {
24402520
use lightning::util::ser::{Readable, Writeable};

src/liquidity/mod.rs

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -517,6 +517,16 @@ where
517517
select_lsps_for_protocol(&self.lsp_nodes, protocol, Some(node_id))
518518
}
519519

520+
pub(crate) fn get_undiscovered_lsps(&self) -> Vec<(PublicKey, SocketAddress)> {
521+
self.lsp_nodes
522+
.read()
523+
.expect("lock")
524+
.iter()
525+
.filter(|n| n.supported_protocols.is_none())
526+
.map(|n| (n.node_id, n.address.clone()))
527+
.collect()
528+
}
529+
520530
/// Flips the `discovery_done` watch to `true`.
521531
///
522532
/// Called once after the *initial* batch of LSPs configured at build time has been

0 commit comments

Comments
 (0)