From 85622083f80a87c76bbcfb75f2c038e660b6e4c2 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 13:41:33 +0100 Subject: [PATCH 1/4] Publish peer store changes only once persisted Adding a peer inserted it into the in-memory set before the write was awaited. If the write failed, a retry found the peer already present and returned success without writing again, so the peer was reported as stored but was gone after a restart. Removing a peer already persisted first, but did so through separate hand-rolled logic. Route both mutations through a single helper that applies the change to a copy of the peer set, persists it, and only then publishes it to readers. The in-memory set can thereby no longer get ahead of the persisted one, and a retry after a failed write persists again. Co-Authored-By: HAL 9000 --- src/peer_store.rs | 74 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 50 insertions(+), 24 deletions(-) diff --git a/src/peer_store.rs b/src/peer_store.rs index 8345bf7111..a9eb1981e6 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -43,45 +43,48 @@ where } pub(crate) async fn add_peer(&self, peer_info: PeerInfo) -> Result<(), Error> { - let _guard = self.mutation_lock.lock().await; - let data = { - let mut locked_peers = self.peers.write().expect("lock"); - if locked_peers.contains_key(&peer_info.node_id) { - return Ok(()); + self.update_peers(|peers| { + if peers.contains_key(&peer_info.node_id) { + return false; } - locked_peers.insert(peer_info.node_id, peer_info); - PeerStoreSerWrapper(&locked_peers).encode() - }; - self.persist_peers(data).await + peers.insert(peer_info.node_id, peer_info); + true + }) + .await } pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { + self.update_peers(|peers| { + peers.remove(node_id); + true + }) + .await + } + + /// Applies `update` to a copy of the peer set, persists the result, and only then publishes + /// it to readers, so that the in-memory peer set never gets ahead of the persisted one. + /// + /// `update` returns whether it changed anything; if it didn't, nothing is written. + async fn update_peers) -> bool>( + &self, update: F, + ) -> 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 updated_peers = self.peers.read().expect("lock").clone(); + if !update(&mut updated_peers) { + return Ok(()); + } + let data = PeerStoreSerWrapper(&updated_peers).encode(); self.persist_peers(data).await?; - self.peers.write().expect("lock").remove(node_id); + *self.peers.write().expect("lock") = updated_peers; Ok(()) } /// Returns the current in-memory peer set. - /// - /// The async mutation lock serializes `add_peer` and `remove_peer`, 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 { self.peers.read().expect("lock").values().cloned().collect() } /// Returns the current in-memory peer info for `node_id`. - /// - /// The async mutation lock serializes `add_peer` and `remove_peer`, 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 { self.peers.read().expect("lock").get(node_id).cloned() } @@ -219,6 +222,16 @@ mod tests { } } + fn test_peer_info() -> PeerInfo { + PeerInfo { + node_id: PublicKey::from_str( + "0276607124ebe6a6c9338517b6f485825b27c2dcc0b9fc2aa6a4c0df91194e5993", + ) + .unwrap(), + address: SocketAddress::from_str("127.0.0.1:9738").unwrap(), + } + } + #[tokio::test] async fn peer_info_persistence() { let store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); @@ -277,4 +290,17 @@ 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)); } + + #[tokio::test] + async fn failed_add_peer_is_not_published_and_retried() { + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let peer_store = PeerStore::new(store, Arc::new(TestLogger::new())); + let peer_info = test_peer_info(); + + assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info.clone()).await); + assert_eq!(None, peer_store.get_peer(&peer_info.node_id)); + + // A retry has to attempt the write again. + assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info).await); + } } From 8544da8473e39c0b3b6e4b4de9d0b5970afc3be3 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 13:41:49 +0100 Subject: [PATCH 2/4] Skip the write when removing an unknown peer Removing a peer rewrote the persisted peer set even if the peer was not stored. With a failing store, callers were thereby handed a persistence error for a removal that had nothing to remove, e.g., when a channel to a peer we never stored was closed. Report whether the removal changed the peer set, so that nothing is written, and nothing can fail, if the peer is unknown. Co-Authored-By: HAL 9000 --- src/peer_store.rs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/peer_store.rs b/src/peer_store.rs index a9eb1981e6..7a191ae671 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -54,11 +54,7 @@ where } pub(crate) async fn remove_peer(&self, node_id: &PublicKey) -> Result<(), Error> { - self.update_peers(|peers| { - peers.remove(node_id); - true - }) - .await + self.update_peers(|peers| peers.remove(node_id).is_some()).await } /// Applies `update` to a copy of the peer set, persists the result, and only then publishes @@ -303,4 +299,13 @@ mod tests { // A retry has to attempt the write again. assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info).await); } + + #[tokio::test] + async fn removing_unknown_peer_does_not_write() { + // Every write to this store fails, so removing only succeeds if nothing is written. + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let peer_store = PeerStore::new(store, Arc::new(TestLogger::new())); + + assert_eq!(Ok(()), peer_store.remove_peer(&test_peer_info().node_id).await); + } } From c9343aa62ce0c18e15f1b404cab8a7cfeb8f3b68 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 13:42:01 +0100 Subject: [PATCH 3/4] Use CollectionLength for the persisted peer count The number of persisted peers was written as a plain `u16`, so a peer set of more than 65535 entries had its count truncated and most of its peers silently dropped on the next restart. The event queue had the same issue, which was fixed by switching to `CollectionLength`. Do the same for the peer store. As `CollectionLength` encodes counts below `u16::MAX` exactly as before, previously persisted peer sets are still read and smaller ones are still written unchanged. As the count read from disk is thereby no longer bounded to `u16::MAX`, we also stop preallocating for it, so that a corrupted count fails to read rather than exhausting memory. Co-Authored-By: HAL 9000 --- src/peer_store.rs | 36 +++++++++++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/src/peer_store.rs b/src/peer_store.rs index 7a191ae671..4a025a7334 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -12,7 +12,7 @@ use std::sync::{Arc, RwLock}; use bitcoin::secp256k1::PublicKey; use lightning::impl_writeable_tlv_based; use lightning::util::persist::KVStore; -use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer}; +use lightning::util::ser::{CollectionLength, Readable, ReadableArgs, Writeable, Writer}; use crate::io::{ PEER_INFO_PERSISTENCE_KEY, PEER_INFO_PERSISTENCE_PRIMARY_NAMESPACE, @@ -132,9 +132,11 @@ impl Readable for PeerStoreDeserWrapper { fn read( reader: &mut R, ) -> Result { - let len: u16 = Readable::read(reader)?; - let mut peers = HashMap::with_capacity(len as usize); - for _ in 0..len { + let len: CollectionLength = Readable::read(reader)?; + // We don't preallocate, as a corrupted count would otherwise have us allocate up to + // `u64::MAX` entries before failing to read them. + let mut peers = HashMap::new(); + for _ in 0..len.0 { let k: PublicKey = Readable::read(reader)?; let v: PeerInfo = Readable::read(reader)?; peers.insert(k, v); @@ -147,7 +149,7 @@ pub(crate) struct PeerStoreSerWrapper<'a>(&'a HashMap); impl Writeable for PeerStoreSerWrapper<'_> { fn write(&self, writer: &mut W) -> Result<(), lightning::io::Error> { - (self.0.len() as u16).write(writer)?; + CollectionLength(self.0.len() as u64).write(writer)?; for (k, v) in self.0.iter() { k.write(writer)?; v.write(writer)?; @@ -308,4 +310,28 @@ mod tests { assert_eq!(Ok(()), peer_store.remove_peer(&test_peer_info().node_id).await); } + + #[test] + fn peer_store_keeps_legacy_encoding_below_u16_max() { + let peer_info = test_peer_info(); + let mut peers = HashMap::new(); + peers.insert(peer_info.node_id, peer_info.clone()); + + // Before `CollectionLength`, the count was written as a plain `u16`. + let mut legacy_bytes = Vec::new(); + (peers.len() as u16).write(&mut legacy_bytes).unwrap(); + peer_info.node_id.write(&mut legacy_bytes).unwrap(); + peer_info.write(&mut legacy_bytes).unwrap(); + + assert_eq!(PeerStoreSerWrapper(&peers).encode(), legacy_bytes); + let read_peers: PeerStoreDeserWrapper = Readable::read(&mut &legacy_bytes[..]).unwrap(); + assert_eq!(read_peers.0, peers); + } + + #[test] + fn peer_store_read_rejects_bogus_peer_count() { + // A count of 2^60 + 0xffff peers, followed by no peers at all. + let bytes = [0xff, 0xff, 0x10, 0, 0, 0, 0, 0, 0, 0]; + assert!(::read(&mut &bytes[..]).is_err()); + } } From 1497cf44d9abddf064f4c536804ea8538bbafd24 Mon Sep 17 00:00:00 2001 From: Elias Rohrer Date: Thu, 12 Nov 2026 13:42:17 +0100 Subject: [PATCH 4/4] Rewrite the peer set after an unconfirmed write A peer store write that fails or is cancelled may still have reached the store, e.g., if a remote store timed out after applying it. The persisted peer set could then differ from the in-memory one, and as mutations that don't change the in-memory set skip the write, the difference was only repaired by the next unrelated change. A peer the user removed could thereby be reconnected to after a restart. Remember that the outcome of a write is unknown until one succeeds, and write the peer set on the next mutation regardless of whether it changes anything. Co-Authored-By: HAL 9000 --- src/peer_store.rs | 33 +++++++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/src/peer_store.rs b/src/peer_store.rs index 4a025a7334..db5809f1f3 100644 --- a/src/peer_store.rs +++ b/src/peer_store.rs @@ -27,7 +27,9 @@ where L::Target: LdkLogger, { peers: RwLock>, - mutation_lock: tokio::sync::Mutex<()>, + // Serializes mutations. The guarded flag is set while the persisted peer set may differ from + // the in-memory one, i.e., once a write was started that we don't know to have succeeded. + mutation_lock: tokio::sync::Mutex, kv_store: Arc, logger: L, } @@ -38,7 +40,7 @@ where { pub(crate) fn new(kv_store: Arc, logger: L) -> Self { let peers = RwLock::new(HashMap::new()); - let mutation_lock = tokio::sync::Mutex::new(()); + let mutation_lock = tokio::sync::Mutex::new(false); Self { peers, mutation_lock, kv_store, logger } } @@ -60,17 +62,22 @@ where /// Applies `update` to a copy of the peer set, persists the result, and only then publishes /// it to readers, so that the in-memory peer set never gets ahead of the persisted one. /// - /// `update` returns whether it changed anything; if it didn't, nothing is written. + /// `update` returns whether it changed anything. If it didn't, nothing is written, unless an + /// earlier write may have left the persisted peer set out of sync. async fn update_peers) -> bool>( &self, update: F, ) -> Result<(), Error> { - let _guard = self.mutation_lock.lock().await; + let mut persisted_out_of_sync = self.mutation_lock.lock().await; let mut updated_peers = self.peers.read().expect("lock").clone(); - if !update(&mut updated_peers) { + let changed = update(&mut updated_peers); + if !changed && !*persisted_out_of_sync { return Ok(()); } let data = PeerStoreSerWrapper(&updated_peers).encode(); + // A write that fails or is cancelled may still have reached the store. + *persisted_out_of_sync = true; self.persist_peers(data).await?; + *persisted_out_of_sync = false; *self.peers.write().expect("lock") = updated_peers; Ok(()) } @@ -120,7 +127,7 @@ where let (kv_store, logger) = args; let read_peers: PeerStoreDeserWrapper = Readable::read(reader)?; let peers: RwLock> = RwLock::new(read_peers.0); - let mutation_lock = tokio::sync::Mutex::new(()); + let mutation_lock = tokio::sync::Mutex::new(false); Ok(Self { peers, mutation_lock, kv_store, logger }) } } @@ -334,4 +341,18 @@ mod tests { let bytes = [0xff, 0xff, 0x10, 0, 0, 0, 0, 0, 0, 0]; assert!(::read(&mut &bytes[..]).is_err()); } + + #[tokio::test] + async fn unconfirmed_write_is_followed_by_a_rewrite() { + let store: Arc = Arc::new(DynStoreWrapper(FailingStore)); + let peer_store = PeerStore::new(store, Arc::new(TestLogger::new())); + let peer_info = test_peer_info(); + + // A write reported as failed may still have reached the store. + assert_eq!(Err(Error::PersistenceFailed), peer_store.add_peer(peer_info.clone()).await); + + // Removing the peer doesn't change the in-memory set, but as we can't tell what the + // store holds, it has to be written regardless. + assert_eq!(Err(Error::PersistenceFailed), peer_store.remove_peer(&peer_info.node_id).await); + } }