Remove force-close peers after reconnect

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],

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

@@ -1702,10 +1702,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_any_node_has_onchain_tx_type(
  &[("node_a", &node_a), ("node_b", &node_b)],