Merge pull request #983 from tnull/2026-07-remove-force-close-peers-after-reconnect

Remove force-close peers after reconnect

Author: tnull

src/connection.rs

@@ -53,6 +53,10 @@ where
 		self.do_connect_peer(node_id, addr).await
 	}
 
+	pub(crate) fn disconnect_peer(&self, node_id: PublicKey) {
+		self.peer_manager.disconnect_by_node_id(node_id);
+	}
+
 	pub(crate) async fn do_connect_peer(
 		&self, node_id: PublicKey, addr: SocketAddress,
 	) -> Result<(), Error> {

src/event.rs

@@ -34,7 +34,7 @@ use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
 use lightning_liquidity::lsps2::utils::compute_opening_fee;
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
 
-use crate::config::{may_announce_channel, Config};
+use crate::config::{may_announce_channel, Config, PEER_RECONNECTION_INTERVAL};
 use crate::connection::ConnectionManager;
 use crate::data_store::DataStoreUpdateResult;
 use crate::fee_estimator::ConfirmationTarget;
@@ -583,6 +583,68 @@ where
 		}
 	}
 
+	fn remove_peer_after_reconnect(&self, peer_info: PeerInfo, closed_channel_id: ChannelId) {
+		let channel_manager = Arc::clone(&self.channel_manager);
+		let connection_manager = Arc::clone(&self.connection_manager);
+		let peer_store = Arc::clone(&self.peer_store);
+		let logger = self.logger.clone();
+		self.runtime.spawn_cancellable_background_task(async move {
+			let has_other_channels = || {
+				channel_manager
+					.list_channels_with_counterparty(&peer_info.node_id)
+					.iter()
+					.any(|c| c.channel_id != closed_channel_id)
+			};
+
+			if peer_store.get_peer(&peer_info.node_id).is_none() || has_other_channels() {
+				return;
+			}
+
+			// Ensure a connected peer cannot be mistaken for a completed recovery reconnect.
+			// With no other channels left, reconnecting once gives `channel_reestablish` a chance
+			// to retransmit the force-close error before we stop persisting the peer.
+			connection_manager.disconnect_peer(peer_info.node_id);
+
+			loop {
+				if peer_store.get_peer(&peer_info.node_id).is_none() || has_other_channels() {
+					return;
+				}
+
+				match connection_manager
+					.connect_peer_if_necessary(peer_info.node_id, peer_info.address.clone())
+					.await
+				{
+					Ok(()) => {
+						if peer_store.get_peer(&peer_info.node_id).is_none() || has_other_channels()
+						{
+							return;
+						}
+						if let Err(e) = peer_store.remove_peer(&peer_info.node_id).await {
+							log_error!(
+								logger,
+								"Failed to remove peer {} from peer store: {}",
+								peer_info.node_id,
+								e
+							);
+						} else {
+							return;
+						}
+					},
+					Err(e) => {
+						log_debug!(
+							logger,
+							"Failed to reconnect peer {} before removing from peer store: {}",
+							peer_info.node_id,
+							e
+						);
+					},
+				}
+
+				tokio::time::sleep(PEER_RECONNECTION_INTERVAL).await;
+			}
+		});
+	}
+
 	async fn fail_claimable_payment(
 		&self, payment_id: PaymentId, payment_hash: &PaymentHash,
 	) -> Result<(), ReplayEvent> {
@@ -1627,25 +1689,24 @@ where
 				let counterparty_node_id = counterparty_node_id
 					.expect("counterparty_node_id is always set since LDK 0.0.117");
 
-				// Drop the peer once its last channel with us has reached a terminal state
-				// that reconnection cannot recover. Every closure reason is terminal except
-				// `HolderForceClosed`: when *we* force-close, we keep reconnecting so that
-				// `channel_reestablish` can drive recovery (see `Node::close_channel_internal`).
+				// Drop the peer once its last channel with us has reached a terminal state.
+				// For `HolderForceClosed`, retain it through one recovery reconnect so that
+				// `channel_reestablish` can retransmit the force-close error before cleanup.
 				// This also cleans up peers persisted for a channel that closed before funding
 				// (e.g. `CounterpartyCoopClosedUnfundedChannel`), which would otherwise be
 				// retried forever.
 				// We exclude `channel_id` from the count because LDK emits `ChannelClosed`
 				// before removing it from its internal list.
-				let dont_reconnect = !matches!(reason, ClosureReason::HolderForceClosed { .. });
-
-				if dont_reconnect {
-					let has_other_channels = self
-						.channel_manager
-						.list_channels_with_counterparty(&counterparty_node_id)
-						.iter()
-						.any(|c| c.channel_id != channel_id);
-
-					if !has_other_channels {
+				let has_other_channels = self
+					.channel_manager
+					.list_channels_with_counterparty(&counterparty_node_id)
+					.iter()
+					.any(|c| c.channel_id != channel_id);
+
+				let peer_to_reconnect = if !has_other_channels {
+					if matches!(reason, ClosureReason::HolderForceClosed { .. }) {
+						self.peer_store.get_peer(&counterparty_node_id)
+					} else {
 						if let Err(e) = self.peer_store.remove_peer(&counterparty_node_id).await {
 							log_error!(
 								self.logger,
@@ -1655,8 +1716,11 @@ where
 							);
 							return Err(ReplayEvent());
 						}
+						None
 					}
-				}
+				} else {
+					None
+				};
 
 				let event = Event::ChannelClosed {
 					channel_id,
@@ -1672,6 +1736,10 @@ where
 						return Err(ReplayEvent());
 					},
 				};
+
+				if let Some(peer_info) = peer_to_reconnect {
+					self.remove_peer_after_reconnect(peer_info, channel_id);
+				}
 			},
 			LdkEvent::DiscardFunding { channel_id, funding_info } => {
 				if let FundingInfo::Contribution { inputs: _, outputs } = funding_info {

src/lib.rs

@@ -2030,12 +2030,10 @@ impl Node {
 			}
 
 			// Peer store cleanup is handled centrally in the `ChannelClosed` event handler,
-			// which drops the peer once its last channel reaches a terminal state that
-			// reconnection cannot recover. We intentionally do nothing here so that a
-			// force-closed peer is retained, letting the background reconnection task keep
-			// firing and drive the `channel_reestablish` recovery flow. This is especially
-			// important against LND peers, which don't always handle force-closure error
-			// messages correctly.
+			// which retains a force-closed peer through one recovery reconnect before
+			// dropping it. This lets `channel_reestablish` drive the recovery flow, which is
+			// especially important against LND peers that don't always handle force-closure
+			// error messages correctly.
 		}
 
 		Ok(())

src/peer_store.rs

@@ -58,11 +58,14 @@ where
 	pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> {
 		let _guard = self.mutation_lock.lock().await;
 		let data = {
-			let mut locked_peers = self.peers.write().expect("lock");
-			locked_peers.remove(node_id);
-			PeerStoreSerWrapper(&locked_peers).encode()
+			let locked_peers = self.peers.read().expect("lock");
+			let mut updated_peers = locked_peers.clone();
+			updated_peers.remove(node_id);
+			PeerStoreSerWrapper(&updated_peers).encode()
 		};
-		self.persist_peers(data).await
+		self.persist_peers(data).await?;
+		self.peers.write().expect("lock").remove(node_id);
+		Ok(())
 	}
 
 	/// Returns the current in-memory peer set.
@@ -170,12 +173,52 @@ mod tests {
 	use std::str::FromStr;
 	use std::sync::Arc;
 
+	use bitcoin::io;
+	use lightning::util::persist::{PageToken, PaginatedKVStore, PaginatedListResponse};
 	use lightning::util::test_utils::TestLogger;
 
 	use super::*;
 	use crate::io::test_utils::InMemoryStore;
 	use crate::types::DynStoreWrapper;
 
+	struct FailingStore;
+
+	impl KVStore for FailingStore {
+		fn read(
+			&self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str,
+		) -> impl std::future::Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send {
+			async { Err(io::Error::new(io::ErrorKind::Other, "read failed")) }
+		}
+
+		fn write(
+			&self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str, _buf: Vec<u8>,
+		) -> impl std::future::Future<Output = Result<(), io::Error>> + 'static + Send {
+			async { Err(io::Error::new(io::ErrorKind::Other, "write failed")) }
+		}
+
+		fn remove(
+			&self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str, _lazy: bool,
+		) -> impl std::future::Future<Output = Result<(), io::Error>> + 'static + Send {
+			async { Err(io::Error::new(io::ErrorKind::Other, "remove failed")) }
+		}
+
+		fn list(
+			&self, _primary_namespace: &str, _secondary_namespace: &str,
+		) -> impl std::future::Future<Output = Result<Vec<String>, io::Error>> + 'static + Send {
+			async { Err(io::Error::new(io::ErrorKind::Other, "list failed")) }
+		}
+	}
+
+	impl PaginatedKVStore for FailingStore {
+		fn list_paginated(
+			&self, _primary_namespace: &str, _secondary_namespace: &str,
+			_page_token: Option<PageToken>,
+		) -> impl std::future::Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + Send
+		{
+			async { Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")) }
+		}
+	}
+
 	#[tokio::test]
 	async fn peer_info_persistence() {
 		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
@@ -215,4 +258,23 @@ mod tests {
 		assert_eq!(peers[0], expected_peer_info);
 		assert_eq!(deser_peer_store.get_peer(&node_id), Some(expected_peer_info));
 	}
+
+	#[tokio::test]
+	async fn remove_peer_does_not_mutate_memory_if_persist_fails() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(FailingStore));
+		let logger = Arc::new(TestLogger::new());
+		let node_id = PublicKey::from_str(
+			"0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993",
+		)
+		.unwrap();
+		let peer_info =
+			PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() };
+		let mut peers = HashMap::new();
+		peers.insert(node_id, peer_info.clone());
+		let persisted_bytes = PeerStoreSerWrapper(&peers).encode();
+		let peer_store = PeerStore::read(&mut &persisted_bytes[..], (store, logger)).unwrap();
+
+		assert_eq!(Err(Error::PersistenceFailed), peer_store.remove_peer(&node_id).await);
+		assert_eq!(Some(peer_info), peer_store.get_peer(&node_id));
+	}
 }

tests/common/mod.rs

@@ -1699,10 +1699,11 @@ pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
 	}
 
 	if force_close {
-		// Peer retained after local force-close to allow channel_reestablish recovery.
+		// The recovery reconnect completed while the force-close settled, so the peer no longer
+		// needs to remain persisted.
 		assert!(
-			node_a.list_peers().iter().any(|p| p.node_id == node_b.node_id() && p.is_persisted),
-			"node_b should remain persisted in node_a peer store after locally-initiated force-close"
+			!node_a.list_peers().iter().any(|p| p.node_id == node_b.node_id() && p.is_persisted),
+			"node_b should be removed from node_a peer store after the recovery reconnect"
 		);
 		assert_all_nodes_have_onchain_tx_type(
 			&[("node_a", &node_a), ("node_b", &node_b)],