Skip to content

Commit 32d5f02

Browse files
committed
fix(peer_store): upsert peer address when re-adding known peer
Previously `add_peer` returned early via `contains_key`, so re-adding a known peer with a new SocketAddress was silently dropped. Now we upsert the address/token and persist before mutating in-memory state, matching `remove_peer`'s write-then-commit pattern. Fixes #700. (The liquidity-side upsert was already merged upstream in a08c5f7; this PR keeps the peer_store store-persistence fix that issue #700 also requires.) AI-assisted (Sera). Signed-off-by: Bartok9 <259807879+Bartok9@users.noreply.github.com>
1 parent a08c5f7 commit 32d5f02

1 file changed

Lines changed: 121 additions & 5 deletions

File tree

‎src/peer_store.rs‎

Lines changed: 121 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,17 +42,44 @@ where
4242
Self { peers, mutation_lock, kv_store, logger }
4343
}
4444

45+
/// Inserts or updates a peer entry.
46+
///
47+
/// If the peer is already known with the same address, this is a no-op. If the peer is new or
48+
/// the stored address changed (e.g. an LSP migrated hosts), the entry is updated and persisted.
49+
/// In-memory state is only mutated after a successful store write, matching [`Self::remove_peer`].
4550
pub(crate) async fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> {
4651
let _guard = self.mutation_lock.lock().await;
4752
let data = {
4853
let mut locked_peers = self.peers.write().expect("lock");
49-
if locked_peers.contains_key(&peer_info.node_id) {
50-
return Ok(());
54+
if let Some(existing) = locked_peers.get(&peer_info.node_id) {
55+
if existing.address == peer_info.address {
56+
return Ok(());
57+
}
58+
}
59+
60+
// Temporarily apply the update so it is included in the
61+
// serialized representation.
62+
let previous = locked_peers.insert(peer_info.node_id, peer_info.clone());
63+
let data = PeerStoreSerWrapper(&locked_peers).encode();
64+
65+
// Restore the original in-memory state before persistence.
66+
match previous {
67+
Some(previous) => {
68+
locked_peers.insert(peer_info.node_id, previous);
69+
},
70+
None => {
71+
locked_peers.remove(&peer_info.node_id);
72+
},
5173
}
52-
locked_peers.insert(peer_info.node_id, peer_info);
53-
PeerStoreSerWrapper(&locked_peers).encode()
74+
75+
data
5476
};
55-
self.persist_peers(data).await
77+
78+
self.persist_peers(data).await?;
79+
80+
// Persistence succeeded, so apply the update permanently.
81+
self.peers.write().expect("lock").insert(peer_info.node_id, peer_info);
82+
Ok(())
5683
}
5784

5885
pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> {
@@ -277,4 +304,93 @@ mod tests {
277304
assert_eq!(Err(Error::PersistenceFailed), peer_store.remove_peer(&node_id).await);
278305
assert_eq!(Some(peer_info), peer_store.get_peer(&node_id));
279306
}
307+
308+
#[tokio::test]
309+
async fn peer_address_updated_on_readd() {
310+
let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
311+
let logger = Arc::new(TestLogger::new());
312+
let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger));
313+
314+
let node_id = PublicKey::from_str(
315+
"0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993",
316+
)
317+
.unwrap();
318+
let old_address = SocketAddress::from_str("127.0.0.1:9738").unwrap();
319+
let new_address = SocketAddress::from_str("127.0.0.1:9739").unwrap();
320+
321+
peer_store.add_peer(PeerInfo { node_id, address: old_address.clone() }).await.unwrap();
322+
assert_eq!(peer_store.get_peer(&node_id), Some(PeerInfo { node_id, address: old_address }));
323+
324+
// Re-adding the same peer with a new socket address must refresh the stored entry
325+
// (regression for https://github.com/lightningdevkit/ldk-node/issues/700).
326+
let updated = PeerInfo { node_id, address: new_address.clone() };
327+
peer_store.add_peer(updated.clone()).await.unwrap();
328+
assert_eq!(peer_store.get_peer(&node_id), Some(updated.clone()));
329+
330+
let persisted_bytes = KVStore::read(
331+
&*store,
332+
PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
333+
PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
334+
PEER_INFO_PERSISTENCE_KEY,
335+
)
336+
.await
337+
.unwrap();
338+
let deser_peer_store =
339+
PeerStore::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
340+
assert_eq!(deser_peer_store.get_peer(&node_id), Some(updated));
341+
}
342+
343+
#[tokio::test]
344+
async fn peer_same_address_skips_persist() {
345+
let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
346+
let logger = Arc::new(TestLogger::new());
347+
let peer_store = PeerStore::new(Arc::clone(&store), Arc::clone(&logger));
348+
349+
let node_id = PublicKey::from_str(
350+
"0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993",
351+
)
352+
.unwrap();
353+
let address = SocketAddress::from_str("127.0.0.1:9738").unwrap();
354+
let peer_info = PeerInfo { node_id, address };
355+
356+
peer_store.add_peer(peer_info.clone()).await.unwrap();
357+
let first_bytes = KVStore::read(
358+
&*store,
359+
PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
360+
PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
361+
PEER_INFO_PERSISTENCE_KEY,
362+
)
363+
.await
364+
.unwrap();
365+
366+
// Identical re-add is a no-op for the store payload.
367+
peer_store.add_peer(peer_info.clone()).await.unwrap();
368+
let second_bytes = KVStore::read(
369+
&*store,
370+
PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
371+
PEER_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
372+
PEER_INFO_PERSISTENCE_KEY,
373+
)
374+
.await
375+
.unwrap();
376+
assert_eq!(first_bytes, second_bytes);
377+
assert_eq!(peer_store.get_peer(&node_id), Some(peer_info));
378+
}
379+
380+
#[tokio::test]
381+
async fn add_peer_does_not_mutate_memory_if_persist_fails() {
382+
let store: Arc<DynStore> = Arc::new(DynStoreWrapper(FailingStore));
383+
let logger = Arc::new(TestLogger::new());
384+
let peer_store = PeerStore::new(store, logger);
385+
386+
let node_id = PublicKey::from_str(
387+
"0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993",
388+
)
389+
.unwrap();
390+
let peer_info =
391+
PeerInfo { node_id, address: SocketAddress::from_str("127.0.0.1:9738").unwrap() };
392+
393+
assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info.clone()).await);
394+
assert_eq!(None, peer_store.get_peer(&node_id));
395+
}
280396
}

0 commit comments

Comments
 (0)