From 1f08ef1f6d41ac616dc632a374daffe706f67072 Mon Sep 17 00:00:00 2001 From: parthod0x Date: Wed, 28 Oct 2026 19:28:15 +0000 Subject: [PATCH] Fix u16 truncation of persisted event queue and peer store (#1138) `EventQueueSerWrapper` and `PeerStoreSerWrapper` wrote their collection length with `(len as u16)` over unbounded collections. Once a queue held more than 65535 entries the persisted count wrapped (65_536 events wrote a count of 0), so on restart the reader restored a truncated queue and application event notifications were silently lost. Use `lightning::util::ser::CollectionLength` for the length prefix in both the serialization and deserialization wrappers, matching how LDK encodes its own collections. This preserves the existing two-byte encoding for counts below 65535 and extends to larger counts. Add round-trip tests at 65534/65535/65536/65540 entries and a legacy two-byte length-prefix decode test to lock in backward compatibility. --- src/event.rs | 75 +++++++++++++++++++++++++++++++++++++++++++++-- src/peer_store.rs | 6 ++-- 2 files changed, 75 insertions(+), 6 deletions(-) diff --git a/src/event.rs b/src/event.rs index 99b217fd52..e974eef9dd 100644 --- a/src/event.rs +++ b/src/event.rs @@ -32,7 +32,7 @@ use lightning::sign::{EntropySource, SpendableOutputDescriptor}; use lightning::util::config::{ChannelConfigOverrides, ChannelConfigUpdate}; use lightning::util::errors::APIError; use lightning::util::persist::KVStore; -use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer}; +use lightning::util::ser::{CollectionLength, Readable, ReadableArgs, Writeable, Writer}; 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}; @@ -519,7 +519,7 @@ impl Readable for EventQueueDeserWrapper { fn read( reader: &mut R, ) -> Result { - let len: u16 = Readable::read(reader)?; + let len: u64 = CollectionLength::read(reader)?.0; let mut queue = VecDeque::with_capacity(len as usize); for _ in 0..len { queue.push_back(Readable::read(reader)?); @@ -532,7 +532,7 @@ struct EventQueueSerWrapper<'a>(&'a VecDeque); impl Writeable for EventQueueSerWrapper<'_> { 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 e in self.0.iter() { e.write(writer)?; } @@ -2947,6 +2947,75 @@ mod tests { assert_eq!(event_queue.next_event(), None); } + #[test] + fn event_queue_round_trips_across_u16_boundary() { + // Regression test for #1138: the persisted event queue length must not be + // truncated to `u16`. Queues holding `u16::MAX` or more events must round-trip + // without losing entries. On the old `(len as u16)` encoding, a queue of 65_536 + // events writes a length of zero, so this assertion fails. + let template = Event::ChannelReady { + channel_id: ChannelId([7u8; 32]), + user_channel_id: UserChannelId(7), + counterparty_node_id: None, + funding_txo: None, + }; + + for count in [65_534usize, 65_535, 65_536, 65_540] { + let mut queue = VecDeque::with_capacity(count); + for _ in 0..count { + queue.push_back(template.clone()); + } + + let mut bytes = Vec::new(); + EventQueueSerWrapper(&queue).write(&mut bytes).unwrap(); + + let read: EventQueueDeserWrapper = Readable::read(&mut &bytes[..]).unwrap(); + assert_eq!( + read.0.len(), + count, + "a queue of {count} events must round-trip without truncation" + ); + assert_eq!(read.0.front(), Some(&template)); + assert_eq!(read.0.back(), Some(&template)); + } + } + + #[test] + fn event_queue_decodes_legacy_u16_length_prefix() { + // Below 65_535 the `CollectionLength` encoding is byte-identical to the previous + // bare `u16` length prefix, so snapshots written by older versions still decode + // and small queues keep their two-byte length prefix. + let template = Event::ChannelReady { + channel_id: ChannelId([9u8; 32]), + user_channel_id: UserChannelId(9), + counterparty_node_id: None, + funding_txo: None, + }; + let count = 3usize; + + let mut legacy = Vec::new(); + (count as u16).write(&mut legacy).unwrap(); + for _ in 0..count { + template.write(&mut legacy).unwrap(); + } + + let read: EventQueueDeserWrapper = Readable::read(&mut &legacy[..]).unwrap(); + assert_eq!(read.0.len(), count); + assert_eq!(read.0.front(), Some(&template)); + + let mut queue = VecDeque::new(); + for _ in 0..count { + queue.push_back(template.clone()); + } + let mut bytes = Vec::new(); + EventQueueSerWrapper(&queue).write(&mut bytes).unwrap(); + assert_eq!( + &bytes[..2], + &legacy[..2], + "a small queue's length prefix must stay a two-byte encoding" + ); + } + #[derive(Clone, Debug, PartialEq, Eq)] enum LegacyEvent { ChannelClosed { diff --git a/src/peer_store.rs b/src/peer_store.rs index 8345bf7111..df00a2f008 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, @@ -133,7 +133,7 @@ impl Readable for PeerStoreDeserWrapper { fn read( reader: &mut R, ) -> Result { - let len: u16 = Readable::read(reader)?; + let len: u64 = CollectionLength::read(reader)?.0; let mut peers = HashMap::with_capacity(len as usize); for _ in 0..len { let k: PublicKey = Readable::read(reader)?; @@ -148,7 +148,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)?;