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’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cryptokvc wants to merge 2 commits into from
Closed

Use CollectionLength for the persisted event queue length #1145

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
15d0885
Use CollectionLength for the persisted event queue length
cryptokvc Oct 9, 2026
596385e
Merge branch 'main' into patch-1
cryptokvc Nov 10, 2026
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 @@ -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};
Expand Down Expand Up @@ -519,9 +519,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.

@cryptokvc cryptokvc Nov 10, 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.

Thanks for the question, and sorry for the slow reply.

The cap only applies to the initial capacity, not to how many entries get read. The length prefix comes from disk, and with CollectionLength it can be any u64 (before, it was at most 65,535). On a corrupted or truncated file, with_capacity(len) would try to allocate up to len * size_of::<Event>() bytes and abort the process, instead of failing the next Readable::read with a DecodeError. Capping it keeps a bad length on the error path.

When the data really is there, the queue just grows past the cap. That's amortized: going from 1,024 to 65,536 entries is about six reallocations, which is small next to decoding the events themselves. LDK's own collection readers in util::ser do the same thing (cmp::min(len.0 as usize, MAX_BUF_SIZE / size_of::<T>())), and I'm happy to switch to that bound for consistency.

If you'd rather have the exact preallocation, I can also drop the cap. The essential part of the PR is the length encoding: today a queue of 65,536+ events is silently truncated on restart (#1138).

@tnull tnull Nov 10, 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.

The cap only applies to the initial capacity, not to how many entries get read

Yes.

When the data really is there, the queue just grows past the cap.

Yes, but it may reallocate many times, making the situation arguably worse.

If you'd rather have the exact preallocation, I can also drop the cap. The essential part of the PR is the length encoding: today a queue of 65,536+ events is silently truncated on restart (#1138).

Yes, thanks, that would be preferable. We shouldn't just continue with corrupted data. If the length prefix is corrupted the chances are the actual data is to.

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 @@ -532,7 +533,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 @@ -3223,4 +3224,42 @@ mod tests {
assert_eq!(sweepable_outpoints(&outputs), vec![(outpoint(2).txid, 2)]);
assert_eq!(direct_outpoints(&outputs), vec![(outpoint(1).txid, 1)]);
}

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