Skip to content

Use CollectionLength for the persisted event queue length #1145

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cryptokvc wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Use CollectionLength for the persisted event queue length #1145

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 44 additions & 5 deletions src/event.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ use lightning::sign::EntropySource;
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};
Expand Down Expand Up @@ -510,9 +510,10 @@ impl Readable for EventQueueDeserWrapper {
fn read<R: lightning::io::Read>(
reader: &mut R,
) -> Result<Self, lightning::ln::msgs::DecodeError> {
let len: u16 = Readable::read(reader)?;
let mut queue = VecDeque::with_capacity(len as usize);
for _ in 0..len {
let len: CollectionLength = Readable::read(reader)?;
// Bound the preallocation, as the length is read from persisted data.

@tnull tnull Oct 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you comment on why this is useful? If the data is there, we will read all the entries, and hence costly re-allocate?

@cryptokvc cryptokvc Oct 15, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's only there for a corrupted or truncated store. With the old u16 count the worst case was 65,535 slots, but CollectionLength can decode to any u64, so a bad length would make with_capacity abort the process before the first event read fails with a DecodeError. The Vec readers in lightning::util::ser bound it the same way (min(len, MAX_BUF_SIZE / size_of::<T>())). For a valid queue longer than the cap it only costs the usual amortized growth. Happy to switch to that MAX_BUF_SIZE form to match LDK, or to drop the cap if you'd rather keep it simple.

@tnull tnull Oct 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, the thing is if we encounter corrupted data here there is no way to recover anyways. I guess we can bound to the minimum of actual len and u16::MAX, or just leave it out. In general, we don't expect users to ever hit an event queue of 1024 events (in fact if they do, that's very bad), but if they ever did adding additional re-allocations on top is probably not great.

let mut queue = VecDeque::with_capacity(core::cmp::min(len.0, 1024) as usize);
for _ in 0..len.0 {
queue.push_back(Readable::read(reader)?);
}
Ok(Self(queue))
Expand All @@ -523,7 +524,7 @@ struct EventQueueSerWrapper<'a>(&'a VecDeque<Event>);

impl Writeable for EventQueueSerWrapper<'_> {
fn write<W: Writer>(&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)?;
}
Expand Down Expand Up @@ -2635,4 +2636,42 @@ mod tests {
}
assert_eq!(event_queue.next_event(), None);
}

fn test_event_queue(len: usize) -> VecDeque<Event> {
(0..len)
.map(|i| Event::ChannelReady {
channel_id: ChannelId([23u8; 32]),
user_channel_id: UserChannelId(i as u128),
counterparty_node_id: None,
funding_txo: None,
})
.collect()
}

#[test]
fn event_queue_round_trips_around_u16_max() {
for len in [65_534, 65_535, 65_536, 65_540] {
let queue = test_event_queue(len);
let bytes = EventQueueSerWrapper(&queue).encode();
let read_queue: EventQueueDeserWrapper = Readable::read(&mut &bytes[..]).unwrap();
assert_eq!(read_queue.0.len(), len);
assert_eq!(read_queue.0, queue);
}
}

#[test]
fn event_queue_keeps_legacy_encoding_below_u16_max() {
let queue = test_event_queue(3);

// Before `CollectionLength`, the count was written as a plain `u16`.
let mut legacy_bytes = Vec::new();
(queue.len() as u16).write(&mut legacy_bytes).unwrap();
for event in queue.iter() {
event.write(&mut legacy_bytes).unwrap();
}

assert_eq!(EventQueueSerWrapper(&queue).encode(), legacy_bytes);
let read_queue: EventQueueDeserWrapper = Readable::read(&mut &legacy_bytes[..]).unwrap();
assert_eq!(read_queue.0, queue);
}
}
Loading