connection: add per-peer exponential reconnection backoff

src/config.rs

@@ -87,6 +87,9 @@ pub(crate) const LDK_PAYMENT_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
 // The time in-between peer reconnection attempts.
 pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
 
+// The upper bound on the per-peer exponential backoff applied to failed reconnection attempts.
+pub(crate) const PEER_RECONNECTION_MAX_INTERVAL: Duration = Duration::from_secs(60 * 30);
+
 // The time in-between RGS sync attempts.
 pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);
 

src/connection.rs

@@ -8,22 +8,51 @@
 use std::collections::hash_map::{self, HashMap};
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use bitcoin::secp256k1::PublicKey;
 use lightning::ln::msgs::SocketAddress;
 
-use crate::config::TorConfig;
+use crate::config::{TorConfig, PEER_RECONNECTION_INTERVAL, PEER_RECONNECTION_MAX_INTERVAL};
 use crate::logger::{log_debug, log_error, log_info, LdkLogger};
 use crate::types::{KeysManager, PeerManager};
 use crate::Error;
 
+struct PeerReconnectState {
+	consecutive_failures: u32,
+	next_retry_at: Instant,
+	next_backoff: Duration,
+}
+
+impl PeerReconnectState {
+	fn new(now: Instant) -> Self {
+		Self {
+			consecutive_failures: 0,
+			next_retry_at: now,
+			next_backoff: PEER_RECONNECTION_INTERVAL,
+		}
+	}
+
+	/// Bumps the failure count, schedules `next_retry_at` to `now + current backoff`,
+	/// and doubles the backoff for the following failure (capped at
+	/// [`PEER_RECONNECTION_MAX_INTERVAL`]). Returns the backoff that was scheduled.
+	fn record_failure(&mut self, now: Instant) -> Duration {
+		self.consecutive_failures = self.consecutive_failures.saturating_add(1);
+		let scheduled_backoff = self.next_backoff;
+		self.next_retry_at = now + scheduled_backoff;
+		self.next_backoff =
+			std::cmp::min(scheduled_backoff.saturating_mul(2), PEER_RECONNECTION_MAX_INTERVAL);
+		scheduled_backoff
+	}
+}
+
 pub(crate) struct ConnectionManager<L: Deref + Clone + Sync + Send>
 where
 	L::Target: LdkLogger,
 {
 	pending_connections:
 		Mutex<HashMap<PublicKey, Vec<tokio::sync::oneshot::Sender<Result<(), Error>>>>>,
+	reconnect_state: Mutex<HashMap<PublicKey, PeerReconnectState>>,
 	peer_manager: Arc<PeerManager>,
 	tor_proxy_config: Option<TorConfig>,
 	keys_manager: Arc<KeysManager>,
@@ -39,14 +68,81 @@ where
 		keys_manager: Arc<KeysManager>, logger: L,
 	) -> Self {
 		let pending_connections = Mutex::new(HashMap::new());
+		let reconnect_state = Mutex::new(HashMap::new());
 
-		Self { pending_connections, peer_manager, tor_proxy_config, keys_manager, logger }
+		Self {
+			pending_connections,
+			reconnect_state,
+			peer_manager,
+			tor_proxy_config,
+			keys_manager,
+			logger,
+		}
+	}
+
+	/// Returns whether the background reconnection task should attempt to reconnect
+	/// to `node_id` now, based on per-peer exponential backoff state.
+	pub(crate) fn is_reconnect_due(&self, node_id: &PublicKey) -> bool {
+		self.reconnect_state
+			.lock()
+			.expect("lock")
+			.get(node_id)
+			.map_or(true, |state| Instant::now() >= state.next_retry_at)
+	}
+
+	/// Records a failed background reconnection attempt: the per-peer retry interval
+	/// is doubled (up to [`PEER_RECONNECTION_MAX_INTERVAL`]) and the next retry is
+	/// scheduled `now + <previous interval>`. `now` should be the instant of the
+	/// reconnection loop tick that scheduled the attempt, so that the resulting
+	/// schedule stays aligned with the loop's wakeups rather than drifting by the
+	/// attempt's duration.
+	///
+	/// Successful connects clear any backoff state via [`Self::do_connect_peer`].
+	pub(crate) fn record_reconnect_failure(&self, node_id: &PublicKey, now: Instant) {
+		let mut state_lock = self.reconnect_state.lock().expect("lock");
+		let state = state_lock.entry(*node_id).or_insert_with(|| PeerReconnectState::new(now));
+		let scheduled_backoff = state.record_failure(now);
+
+		log_debug!(
+			self.logger,
+			"Reconnection to peer {} failed ({} consecutive failures); next retry in {}s",
+			node_id,
+			state.consecutive_failures,
+			scheduled_backoff.as_secs(),
+		);
+	}
+
+	/// Removes any per-peer backoff state for `node_id`, so a subsequent attempt
+	/// is treated as a fresh first try. Called when a peer is removed from the
+	/// persisted peer store.
+	pub(crate) fn clear_reconnect_state(&self, node_id: &PublicKey) {
+		self.reconnect_state.lock().expect("lock").remove(node_id);
+	}
+
+	/// Drops backoff state for any peer not in `persisted_peers`. This guards against
+	/// entries resurrected by a failed attempt that raced the peer's removal from the
+	/// store, which would otherwise linger forever.
+	pub(crate) fn prune_reconnect_state(&self, persisted_peers: &[PublicKey]) {
+		self.reconnect_state
+			.lock()
+			.expect("lock")
+			.retain(|node_id, _| persisted_peers.contains(node_id));
+	}
+
+	/// Returns whether an outbound connection attempt to `node_id` is currently in
+	/// flight, i.e., whether a [`Self::do_connect_peer`] call would merely subscribe
+	/// to another task's attempt (possibly targeting a different address).
+	pub(crate) fn has_pending_connection(&self, node_id: &PublicKey) -> bool {
+		self.pending_connections.lock().expect("lock").contains_key(node_id)
 	}
 
 	pub(crate) async fn connect_peer_if_necessary(
 		&self, node_id: PublicKey, addr: SocketAddress,
 	) -> Result<(), Error> {
 		if self.peer_manager.peer_by_node_id(&node_id).is_some() {
+			// The peer is demonstrably reachable: reset any backoff so a subsequent
+			// drop is retried promptly.
+			self.clear_reconnect_state(&node_id);
 			return Ok(());
 		}
 
@@ -57,6 +153,11 @@ where
 		&self, node_id: PublicKey, addr: SocketAddress,
 	) -> Result<(), Error> {
 		let res = self.do_connect_peer_internal(node_id, addr).await;
+		if res.is_ok() {
+			// Any successful connect (including user-initiated ones) resets backoff so the
+			// background reconnection loop retries promptly if the peer drops again.
+			self.clear_reconnect_state(&node_id);
+		}
 		self.propagate_result_to_subscribers(&node_id, res);
 		res
 	}
@@ -273,3 +374,49 @@ where
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn reconnect_state_doubles_until_capped() {
+		let start = Instant::now();
+		let mut state = PeerReconnectState::new(start);
+
+		let scheduled = state.record_failure(start);
+		assert_eq!(scheduled, PEER_RECONNECTION_INTERVAL);
+		assert_eq!(state.consecutive_failures, 1);
+		assert_eq!(state.next_retry_at, start + PEER_RECONNECTION_INTERVAL);
+
+		let mut expected = PEER_RECONNECTION_INTERVAL;
+		for failure_count in 2..32 {
+			expected = std::cmp::min(expected.saturating_mul(2), PEER_RECONNECTION_MAX_INTERVAL);
+			let scheduled = state.record_failure(start);
+			assert_eq!(scheduled, expected);
+			assert_eq!(state.consecutive_failures, failure_count);
+			assert_eq!(state.next_retry_at, start + expected);
+			assert!(state.next_backoff <= PEER_RECONNECTION_MAX_INTERVAL);
+		}
+
+		// Once capped, further failures stay at the cap.
+		assert_eq!(state.next_backoff, PEER_RECONNECTION_MAX_INTERVAL);
+		let scheduled = state.record_failure(start);
+		assert_eq!(scheduled, PEER_RECONNECTION_MAX_INTERVAL);
+		assert_eq!(state.next_backoff, PEER_RECONNECTION_MAX_INTERVAL);
+	}
+
+	#[test]
+	fn reconnect_state_schedules_relative_to_failure_time() {
+		let t0 = Instant::now();
+		let mut state = PeerReconnectState::new(t0);
+
+		let _ = state.record_failure(t0);
+		assert_eq!(state.next_retry_at, t0 + PEER_RECONNECTION_INTERVAL);
+
+		let t1 = t0 + Duration::from_secs(5);
+		let scheduled = state.record_failure(t1);
+		assert_eq!(scheduled, PEER_RECONNECTION_INTERVAL * 2);
+		assert_eq!(state.next_retry_at, t1 + PEER_RECONNECTION_INTERVAL * 2);
+	}
+}

src/event.rs

@@ -1655,6 +1655,7 @@ where
 							);
 							return Err(ReplayEvent());
 						}
+						self.connection_manager.clear_reconnect_state(&counterparty_node_id);
 					}
 				}
 

src/lib.rs

@@ -484,11 +484,45 @@ impl Node {
 								.map(|peer| peer.counterparty_node_id)
 								.collect::<Vec<_>>();
 
-							for peer_info in connect_peer_store.list_peers().iter().filter(|info| !pm_peers.contains(&info.node_id)) {
-								let _ = connect_cm.do_connect_peer(
+							let persisted_peers = connect_peer_store.list_peers();
+							let persisted_node_ids = persisted_peers
+								.iter()
+								.map(|peer| peer.node_id)
+								.collect::<Vec<_>>();
+							connect_cm.prune_reconnect_state(&persisted_node_ids);
+
+							// Anchor this tick's backoff bookkeeping to the tick instant
+							// rather than each attempt's completion time, so retries stay
+							// aligned with the loop's wakeups.
+							let tick_start = Instant::now();
+
+							for peer_info in persisted_peers.iter() {
+								if pm_peers.contains(&peer_info.node_id) {
+									// A connected peer (e.g., via an inbound connection) is
+									// proven reachable: reset any backoff so a future
+									// disconnect is retried promptly again. Note we only
+									// observe this at tick time: an inbound connection that
+									// comes and goes entirely within one tick keeps its
+									// backoff.
+									connect_cm.clear_reconnect_state(&peer_info.node_id);
+									continue;
+								}
+								if connect_cm.has_pending_connection(&peer_info.node_id) {
+									// Another task is already dialing this peer, possibly
+									// towards a different address. Don't join it: its result
+									// shouldn't drive our backoff for the persisted address.
+									continue;
+								}
+								if !connect_cm.is_reconnect_due(&peer_info.node_id) {
+									continue;
+								}
+								let res = connect_cm.do_connect_peer(
 									peer_info.node_id,
 									peer_info.address.clone(),
 									).await;
+								if res.is_err() {
+									connect_cm.record_reconnect_failure(&peer_info.node_id, tick_start);
+								}
 							}
 						}
 				}
@@ -1221,6 +1255,7 @@ impl Node {
 				log_error!(self.logger, "Failed to remove peer {}: {}", counterparty_node_id, e)
 			},
 		}
+		self.connection_manager.clear_reconnect_state(&counterparty_node_id);
 
 		self.peer_manager.disconnect_by_node_id(counterparty_node_id);
 		Ok(())