From 1b28e6356f85663817ac7f03e0f33eb3404e2076 Mon Sep 17 00:00:00 2001 From: Vincenzo Palazzo Date: Tue, 13 Oct 2026 16:44:14 -0400 Subject: [PATCH] fix(peer_store): upsert peer address when re-adding known peer PeerStore::add_peer returned early when the pubkey was already stored, so Node::connect(..., persist=true) with a new address silently kept the stale address used for reconnection (see #700). Split add_peer into two methods sharing a persist-then-apply helper, following the maintainer review on #1002: - upsert_peer updates the stored address on re-add. Used where the address is explicitly supplied: Node::connect with persist set, open_channel, and the LSPS2 payment path. - add_peer_if_missing only stores unknown peers. Used for the gossip-derived address of inbound channel counterparties, which must not override an explicitly stored address. In-memory state is only mutated after a successful write, matching remove_peer. Following the review suggestion on #1002, avoid cloning the entire peer map: apply the update temporarily under the write lock for serialization, restore it, persist, then apply it permanently. Readers cannot observe the temporary value as they need the same lock. Re-adding an unchanged peer is a no-op. remove_peer is switched to the same pattern for consistency. Addresses the peer-store side of #700. The liquidity-source side (NodeBuilder/Liquidity add_liquidity_source) is deliberately left out here; it is tracked in #1147. Supersedes #1002 and #1115. Co-authored-by: Bartok <259807879+Bartok9@users.noreply.github.com> Assisted-by: goose (AI) --- src/event.rs | 2 +- src/lib.rs | 6 +- src/payment/bolt11.rs | 2 +- src/peer_store.rs | 210 +++++++++++++++++++++++++++++++++++++++--- 4 files changed, 203 insertions(+), 17 deletions(-) diff --git a/src/event.rs b/src/event.rs index 657c938ed6..e57c1f2d4b 100644 --- a/src/event.rs +++ b/src/event.rs @@ -2049,7 +2049,7 @@ where }) }; if let Some(peer) = peer_to_store { - self.peer_store.add_peer(peer).await.unwrap_or_else(|e| { + self.peer_store.add_peer_if_missing(peer).await.unwrap_or_else(|e| { log_error!( self.logger, "Failed to add peer {} to peer store: {}", diff --git a/src/lib.rs b/src/lib.rs index 83bcc6afa4..8781948a59 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1317,6 +1317,8 @@ impl Node { /// Connect to a node on the peer-to-peer network. /// /// If `persist` is set to `true`, we'll remember the peer and reconnect to it on restart. + /// A later `connect` with `persist` set to `true` and a different address updates the + /// stored address used for reconnection. pub fn connect( &self, node_id: PublicKey, address: SocketAddress, persist: bool, ) -> Result<(), Error> { @@ -1339,7 +1341,7 @@ impl Node { log_info!(self.logger, "Connected to peer {}@{}. ", peer_info.node_id, peer_info.address); if persist { - self.runtime.block_on(self.peer_store.add_peer(peer_info))?; + self.runtime.block_on(self.peer_store.upsert_peer(peer_info))?; } Ok(()) @@ -1477,7 +1479,7 @@ impl Node { zero_reserve_string, peer_info.node_id ); - self.runtime.block_on(self.peer_store.add_peer(peer_info))?; + self.runtime.block_on(self.peer_store.upsert_peer(peer_info))?; Ok(UserChannelId(user_channel_id)) }, Err(e) => { diff --git a/src/payment/bolt11.rs b/src/payment/bolt11.rs index a294912ceb..421f63fa9d 100644 --- a/src/payment/bolt11.rs +++ b/src/payment/bolt11.rs @@ -161,7 +161,7 @@ impl Bolt11Payment { // Persist the chosen LSP peer to make sure we reconnect on restart. let peer_info = PeerInfo { node_id: chosen_lsp.node_id, address: chosen_lsp.address }; - self.runtime.block_on(self.peer_store.add_peer(peer_info))?; + self.runtime.block_on(self.peer_store.upsert_peer(peer_info))?; Ok(invoice) } diff --git a/src/peer_store.rs b/src/peer_store.rs index 8345bf7111..057b22d454 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -42,35 +42,102 @@ where Self { peers, mutation_lock, kv_store, logger } } - pub(crate) async fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { + /// Stores the given peer only if no entry for it exists yet. + /// + /// In particular, an existing peer's stored address is left untouched. Use this for + /// gossip- or event-derived peers (e.g., inbound channel counterparties), where the + /// discovered address must not override an explicitly stored one. + /// + /// The in-memory state is only updated after the new state has been persisted + /// successfully. + pub(crate) async fn add_peer_if_missing(&self, peer_info: PeerInfo) -> Result<(), Error> { + let _guard = self.mutation_lock.lock().await; + + { + let peers = self.peers.read().expect("lock"); + if peers.contains_key(&peer_info.node_id) { + return Ok(()); + } + } + + self.persist_and_apply(peer_info).await + } + + /// Stores the given peer, updating the stored address if the peer is already known. + /// + /// Use this for explicitly supplied addresses (e.g., via [`crate::Node::connect`] with + /// `persist` set), where a re-connect to a new address should replace the one used for + /// reconnection. + /// + /// The in-memory state is only updated after the new state has been persisted + /// successfully. + pub(crate) async fn upsert_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { let _guard = self.mutation_lock.lock().await; + + { + let peers = self.peers.read().expect("lock"); + if peers.get(&peer_info.node_id) == Some(&peer_info) { + return Ok(()); + } + } + + self.persist_and_apply(peer_info).await + } + + async fn persist_and_apply(&self, peer_info: PeerInfo) -> Result<(), Error> { let data = { let mut locked_peers = self.peers.write().expect("lock"); - if locked_peers.contains_key(&peer_info.node_id) { - return Ok(()); + + // Temporarily apply the update so it is included in the serialized state, then + // restore the current state before persisting: readers cannot observe the + // temporary value as they need the same lock. This avoids cloning the entire + // peer map while retaining the persistence-first semantics. + let previous = locked_peers.insert(peer_info.node_id, peer_info.clone()); + let data = PeerStoreSerWrapper(&locked_peers).encode(); + match previous { + Some(previous) => { + locked_peers.insert(peer_info.node_id, previous); + }, + None => { + locked_peers.remove(&peer_info.node_id); + }, } - locked_peers.insert(peer_info.node_id, peer_info); - PeerStoreSerWrapper(&locked_peers).encode() + data }; - self.persist_peers(data).await + + self.persist_peers(data).await?; + + // Persistence succeeded, so apply the update permanently. + self.peers.write().expect("lock").insert(peer_info.node_id, peer_info); + Ok(()) } pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { let _guard = self.mutation_lock.lock().await; let data = { - let locked_peers = self.peers.read().expect("lock"); - let mut updated_peers = locked_peers.clone(); - updated_peers.remove(node_id); - PeerStoreSerWrapper(&updated_peers).encode() + let mut locked_peers = self.peers.write().expect("lock"); + + // Temporarily apply the removal so it is excluded from the serialized state, + // then restore the current state before persisting: readers cannot observe + // the temporary value as they need the same lock. This avoids cloning the + // entire peer map while retaining the persistence-first semantics. + let previous = locked_peers.remove(node_id); + let data = PeerStoreSerWrapper(&locked_peers).encode(); + if let Some(previous) = previous { + locked_peers.insert(*node_id, previous); + } + data }; self.persist_peers(data).await?; + + // Persistence succeeded, so apply the removal permanently. self.peers.write().expect("lock").remove(node_id); Ok(()) } /// Returns the current in-memory peer set. /// - /// The async mutation lock serializes `add_peer` and `remove_peer`, but this synchronous + /// The async mutation lock serializes peer-store mutations, but this synchronous /// reader cannot wait on it. Until peer-store reads are async, callers may observe peer /// changes that are still being persisted. pub(crate) fn list_peers(&self) -> Vec { @@ -79,7 +146,7 @@ where /// Returns the current in-memory peer info for `node_id`. /// - /// The async mutation lock serializes `add_peer` and `remove_peer`, but this synchronous + /// The async mutation lock serializes peer-store mutations, but this synchronous /// reader cannot wait on it. Until peer-store reads are async, callers may observe peer /// changes that are still being persisted. pub(crate) fn get_peer(&self, node_id: &PublicKey) -> Option { @@ -239,7 +306,7 @@ mod tests { ) .await .is_err()); - peer_store.add_peer(expected_peer_info.clone()).await.unwrap(); + peer_store.upsert_peer(expected_peer_info.clone()).await.unwrap(); // Check we can read back what we persisted. let persisted_bytes = KVStore::read( @@ -277,4 +344,121 @@ mod tests { assert_eq!(Err(Error::PersistenceFailed), peer_store.remove_peer(&node_id).await); assert_eq!(Some(peer_info), peer_store.get_peer(&node_id)); } + + fn test_node_id() -> PublicKey { + PublicKey::from_str("0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993") + .unwrap() + } + + #[tokio::test] + async fn upsert_peer_updates_stored_address() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); + + let node_id = test_node_id(); + let old_address = SocketAddress::from_str("127.0.0.1:9738").unwrap(); + let new_address = SocketAddress::from_str("127.0.0.1:9739").unwrap(); + + let old_peer_info = PeerInfo { node_id, address: old_address }; + peer_store.upsert_peer(old_peer_info.clone()).await.unwrap(); + assert_eq!(peer_store.get_peer(&node_id), Some(old_peer_info)); + + // Re-adding the same peer with a new address must update the stored entry, also on + // disk. + let new_peer_info = PeerInfo { node_id, address: new_address }; + peer_store.upsert_peer(new_peer_info.clone()).await.unwrap(); + assert_eq!(peer_store.get_peer(&node_id), Some(new_peer_info.clone())); + + let persisted_bytes = KVStore::read( + &*store, + PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PEER_INFO_PERSISTENCE_KEY, + ) + .await + .unwrap(); + let deser_peer_store = + PeerStore::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap(); + assert_eq!(deser_peer_store.get_peer(&node_id), Some(new_peer_info)); + } + + #[tokio::test] + async fn add_peer_if_missing_does_not_overwrite_stored_address() { + let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let logger = Arc::new(TestLogger::new()); + let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger)); + + let node_id = test_node_id(); + let stored = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() }; + peer_store.upsert_peer(stored.clone()).await.unwrap(); + + // A gossip-/event-derived address must not overwrite the explicitly stored one. + let discovered = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9739").unwrap() }; + peer_store.add_peer_if_missing(discovered).await.unwrap(); + assert_eq!(peer_store.get_peer(&node_id), Some(stored)); + } + + #[tokio::test] + async fn readding_unchanged_peer_skips_persist() { + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let logger = Arc::new(TestLogger::new()); + + let node_id = test_node_id(); + let peer_info = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() }; + + // Seed the store's in-memory state with the peer info. Any attempt to persist + // below would fail, so succeeding proves the no-op paths skip persistence. + 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(); + + // Re-adding the unchanged peer must be a no-op and not write again. + peer_store.upsert_peer(peer_info.clone()).await.unwrap(); + + // Nor must the add-if-missing path write (or overwrite) for a known peer. + let other_peer_info = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9739").unwrap() }; + peer_store.add_peer_if_missing(other_peer_info).await.unwrap(); + assert_eq!(Some(peer_info), peer_store.get_peer(&node_id)); + } + + #[tokio::test] + async fn upsert_peer_does_not_mutate_memory_if_persist_fails() { + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let logger = Arc::new(TestLogger::new()); + + let node_id = test_node_id(); + let old_peer_info = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() }; + let new_peer_info = + PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9739").unwrap() }; + + // Seed the store's in-memory state with the old peer info. + let mut peers = HashMap::new(); + peers.insert(node_id, old_peer_info.clone()); + let persisted_bytes = PeerStoreSerWrapper(&peers).encode(); + let peer_store = + PeerStore::read(&mut &persisted_bytes[..], (Arc::clone(&store), Arc::clone(&logger))) + .unwrap(); + + // Updating the stored address fails, so the old peer info must remain untouched. + assert_eq!( + Err(Error::PersistenceFailed), + peer_store.upsert_peer(new_peer_info.clone()).await + ); + assert_eq!(Some(old_peer_info.clone()), peer_store.get_peer(&node_id)); + + // Inserting a new peer fails too, so no entry must be left behind. + let peer_store = PeerStore::new(store, logger); + assert_eq!( + Err(Error::PersistenceFailed), + peer_store.upsert_peer(old_peer_info.clone()).await + ); + assert_eq!(None, peer_store.get_peer(&node_id)); + } }