From 8847a375f469ba5c9ddc73a9f2a3912f8f203d8e Mon Sep 17 00:00:00 2001 From: Dirkjan Ochtman Date: Wed, 30 Sep 2026 09:40:01 -0700 Subject: [PATCH 1/5] fix(proto): yield error on too many gaps in assembler Port of quinn-rs/quinn@fed0321a (quinn-rs/quinn#2694, RUSTSEC-2026-0185): a peer sending small gapped STREAM or CRYPTO frames could make the assembler buffer an unbounded number of chunks. After defragmenting, more than 1024 chunks now closes the connection with INTERNAL_ERROR. Co-Authored-By: Luke Curley Co-Authored-By: Claude Opus 5.5 --- noq-proto/src/connection/assembler.rs | 159 +++++++++++++---------- noq-proto/src/connection/mod.rs | 4 +- noq-proto/src/connection/streams/recv.rs | 4 +- 3 files changed, 93 insertions(+), 74 deletions(-) diff --git a/noq-proto/src/connection/assembler.rs b/noq-proto/src/connection/assembler.rs index ca0a8d9a06..7156deb039 100644 --- a/noq-proto/src/connection/assembler.rs +++ b/noq-proto/src/connection/assembler.rs @@ -151,7 +151,12 @@ impl Assembler { // Note: If a packet contains many frames from the same stream, the estimated over-allocation // will be much higher because we are counting the same allocation multiple times. - pub(super) fn insert(&mut self, mut offset: u64, mut bytes: Bytes, allocation_size: usize) { + pub(super) fn insert( + &mut self, + mut offset: u64, + mut bytes: Bytes, + allocation_size: usize, + ) -> Result<(), TooManyChunks> { debug_assert!( bytes.len() <= allocation_size, "allocation_size less than bytes.len(): {:?} < {:?}", @@ -180,7 +185,7 @@ impl Assembler { recvd.insert(range); } else if offset < self.bytes_read { if (offset + bytes.len() as u64) <= self.bytes_read { - return; + return Ok(()); } else { let diff = self.bytes_read - offset; offset += diff; @@ -189,7 +194,7 @@ impl Assembler { } if bytes.is_empty() { - return; + return Ok(()); } let buffer = Buffer::new(offset, bytes, allocation_size); self.buffered += buffer.bytes.len(); @@ -210,8 +215,14 @@ impl Assembler { // balance between defragmentation overhead and over-allocation. let threshold = 32768.max(buffered * 3 / 2); if over_allocation > threshold { - self.defragment() + self.defragment(); + // ngtcp2 uses a threshold of 4000 -- try to be a little more conservative? + if self.data.len() > 1024 { + return Err(TooManyChunks); + } } + + Ok(()) } /// Number of bytes consumed by the application @@ -341,6 +352,10 @@ impl State { #[derive(Debug)] pub(crate) struct IllegalOrderedRead; +/// Error indicating that too many chunks are buffered due to maliciously small/gapped frames +#[derive(Debug)] +pub(crate) struct TooManyChunks; + #[cfg(test)] mod test { use super::*; @@ -350,13 +365,13 @@ mod test { fn assemble_ordered() { let mut x = Assembler::new(); assert_matches!(next(&mut x, 32), None); - x.insert(0, Bytes::from_static(b"123"), 3); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); assert_matches!(next(&mut x, 1), Some(ref y) if &y[..] == b"1"); assert_matches!(next(&mut x, 3), Some(ref y) if &y[..] == b"23"); - x.insert(3, Bytes::from_static(b"456"), 3); + x.insert(3, Bytes::from_static(b"456"), 3).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"456"); - x.insert(6, Bytes::from_static(b"789"), 3); - x.insert(9, Bytes::from_static(b"10"), 2); + x.insert(6, Bytes::from_static(b"789"), 3).unwrap(); + x.insert(9, Bytes::from_static(b"10"), 2).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"789"); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"10"); assert_matches!(next(&mut x, 32), None); @@ -366,9 +381,9 @@ mod test { fn assemble_unordered() { let mut x = Assembler::new(); x.ensure_ordering(false).unwrap(); - x.insert(3, Bytes::from_static(b"456"), 3); + x.insert(3, Bytes::from_static(b"456"), 3).unwrap(); assert_matches!(next(&mut x, 32), None); - x.insert(0, Bytes::from_static(b"123"), 3); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123"); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"456"); assert_matches!(next(&mut x, 32), None); @@ -377,8 +392,8 @@ mod test { #[test] fn assemble_duplicate() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"123"), 3); - x.insert(0, Bytes::from_static(b"123"), 3); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123"); assert_matches!(next(&mut x, 32), None); } @@ -386,8 +401,8 @@ mod test { #[test] fn assemble_duplicate_compact() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"123"), 3); - x.insert(0, Bytes::from_static(b"123"), 3); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); x.defragment(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123"); assert_matches!(next(&mut x, 32), None); @@ -396,8 +411,8 @@ mod test { #[test] fn assemble_contained() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"12345"), 5); - x.insert(1, Bytes::from_static(b"234"), 3); + x.insert(0, Bytes::from_static(b"12345"), 5).unwrap(); + x.insert(1, Bytes::from_static(b"234"), 3).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"12345"); assert_matches!(next(&mut x, 32), None); } @@ -405,8 +420,8 @@ mod test { #[test] fn assemble_contained_compact() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"12345"), 5); - x.insert(1, Bytes::from_static(b"234"), 3); + x.insert(0, Bytes::from_static(b"12345"), 5).unwrap(); + x.insert(1, Bytes::from_static(b"234"), 3).unwrap(); x.defragment(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"12345"); assert_matches!(next(&mut x, 32), None); @@ -415,8 +430,8 @@ mod test { #[test] fn assemble_contains() { let mut x = Assembler::new(); - x.insert(1, Bytes::from_static(b"234"), 3); - x.insert(0, Bytes::from_static(b"12345"), 5); + x.insert(1, Bytes::from_static(b"234"), 3).unwrap(); + x.insert(0, Bytes::from_static(b"12345"), 5).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"12345"); assert_matches!(next(&mut x, 32), None); } @@ -424,8 +439,8 @@ mod test { #[test] fn assemble_contains_compact() { let mut x = Assembler::new(); - x.insert(1, Bytes::from_static(b"234"), 3); - x.insert(0, Bytes::from_static(b"12345"), 5); + x.insert(1, Bytes::from_static(b"234"), 3).unwrap(); + x.insert(0, Bytes::from_static(b"12345"), 5).unwrap(); x.defragment(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"12345"); assert_matches!(next(&mut x, 32), None); @@ -434,8 +449,8 @@ mod test { #[test] fn assemble_overlapping() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"123"), 3); - x.insert(1, Bytes::from_static(b"234"), 3); + x.insert(0, Bytes::from_static(b"123"), 3).unwrap(); + x.insert(1, Bytes::from_static(b"234"), 3).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123"); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"4"); assert_matches!(next(&mut x, 32), None); @@ -444,8 +459,8 @@ mod test { #[test] fn assemble_overlapping_compact() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"123"), 4); - x.insert(1, Bytes::from_static(b"234"), 4); + x.insert(0, Bytes::from_static(b"123"), 4).unwrap(); + x.insert(1, Bytes::from_static(b"234"), 4).unwrap(); x.defragment(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"1234"); assert_matches!(next(&mut x, 32), None); @@ -454,10 +469,10 @@ mod test { #[test] fn assemble_complex() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"1"), 1); - x.insert(2, Bytes::from_static(b"3"), 1); - x.insert(4, Bytes::from_static(b"5"), 1); - x.insert(0, Bytes::from_static(b"123456"), 6); + x.insert(0, Bytes::from_static(b"1"), 1).unwrap(); + x.insert(2, Bytes::from_static(b"3"), 1).unwrap(); + x.insert(4, Bytes::from_static(b"5"), 1).unwrap(); + x.insert(0, Bytes::from_static(b"123456"), 6).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123456"); assert_matches!(next(&mut x, 32), None); } @@ -465,10 +480,10 @@ mod test { #[test] fn assemble_complex_compact() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"1"), 1); - x.insert(2, Bytes::from_static(b"3"), 1); - x.insert(4, Bytes::from_static(b"5"), 1); - x.insert(0, Bytes::from_static(b"123456"), 6); + x.insert(0, Bytes::from_static(b"1"), 1).unwrap(); + x.insert(2, Bytes::from_static(b"3"), 1).unwrap(); + x.insert(4, Bytes::from_static(b"5"), 1).unwrap(); + x.insert(0, Bytes::from_static(b"123456"), 6).unwrap(); x.defragment(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"123456"); assert_matches!(next(&mut x, 32), None); @@ -477,19 +492,19 @@ mod test { #[test] fn assemble_old() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"1234"), 4); + x.insert(0, Bytes::from_static(b"1234"), 4).unwrap(); assert_matches!(next(&mut x, 32), Some(ref y) if &y[..] == b"1234"); - x.insert(0, Bytes::from_static(b"1234"), 4); + x.insert(0, Bytes::from_static(b"1234"), 4).unwrap(); assert_matches!(next(&mut x, 32), None); } #[test] fn compact() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"abc"), 4); - x.insert(3, Bytes::from_static(b"def"), 4); - x.insert(9, Bytes::from_static(b"jkl"), 4); - x.insert(12, Bytes::from_static(b"mno"), 4); + x.insert(0, Bytes::from_static(b"abc"), 4).unwrap(); + x.insert(3, Bytes::from_static(b"def"), 4).unwrap(); + x.insert(9, Bytes::from_static(b"jkl"), 4).unwrap(); + x.insert(12, Bytes::from_static(b"mno"), 4).unwrap(); x.defragment(); assert_eq!( next_unordered(&mut x), @@ -504,7 +519,7 @@ mod test { #[test] fn defrag_with_missing_prefix() { let mut x = Assembler::new(); - x.insert(3, Bytes::from_static(b"def"), 3); + x.insert(3, Bytes::from_static(b"def"), 3).unwrap(); x.defragment(); assert_eq!( next_unordered(&mut x), @@ -515,17 +530,17 @@ mod test { #[test] fn defrag_read_chunk() { let mut x = Assembler::new(); - x.insert(3, Bytes::from_static(b"def"), 4); - x.insert(0, Bytes::from_static(b"abc"), 4); - x.insert(7, Bytes::from_static(b"hij"), 4); - x.insert(11, Bytes::from_static(b"lmn"), 4); + x.insert(3, Bytes::from_static(b"def"), 4).unwrap(); + x.insert(0, Bytes::from_static(b"abc"), 4).unwrap(); + x.insert(7, Bytes::from_static(b"hij"), 4).unwrap(); + x.insert(11, Bytes::from_static(b"lmn"), 4).unwrap(); x.defragment(); assert_matches!(x.read(usize::MAX, true), Some(ref y) if &y.bytes[..] == b"abcdef"); - x.insert(5, Bytes::from_static(b"fghijklmn"), 9); + x.insert(5, Bytes::from_static(b"fghijklmn"), 9).unwrap(); assert_matches!(x.read(usize::MAX, true), Some(ref y) if &y.bytes[..] == b"ghijklmn"); - x.insert(13, Bytes::from_static(b"nopq"), 4); + x.insert(13, Bytes::from_static(b"nopq"), 4).unwrap(); assert_matches!(x.read(usize::MAX, true), Some(ref y) if &y.bytes[..] == b"opq"); - x.insert(15, Bytes::from_static(b"pqrs"), 4); + x.insert(15, Bytes::from_static(b"pqrs"), 4).unwrap(); assert_matches!(x.read(usize::MAX, true), Some(ref y) if &y.bytes[..] == b"rs"); assert_matches!(x.read(usize::MAX, true), None); } @@ -534,13 +549,13 @@ mod test { fn unordered_happy_path() { let mut x = Assembler::new(); x.ensure_ordering(false).unwrap(); - x.insert(0, Bytes::from_static(b"abc"), 3); + x.insert(0, Bytes::from_static(b"abc"), 3).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(0, Bytes::from_static(b"abc")) ); assert_eq!(x.read(usize::MAX, false), None); - x.insert(3, Bytes::from_static(b"def"), 3); + x.insert(3, Bytes::from_static(b"def"), 3).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(3, Bytes::from_static(b"def")) @@ -552,15 +567,15 @@ mod test { fn unordered_dedup() { let mut x = Assembler::new(); x.ensure_ordering(false).unwrap(); - x.insert(3, Bytes::from_static(b"def"), 3); + x.insert(3, Bytes::from_static(b"def"), 3).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(3, Bytes::from_static(b"def")) ); assert_eq!(x.read(usize::MAX, false), None); - x.insert(0, Bytes::from_static(b"a"), 1); - x.insert(0, Bytes::from_static(b"abcdefghi"), 9); - x.insert(0, Bytes::from_static(b"abcd"), 4); + x.insert(0, Bytes::from_static(b"a"), 1).unwrap(); + x.insert(0, Bytes::from_static(b"abcdefghi"), 9).unwrap(); + x.insert(0, Bytes::from_static(b"abcd"), 4).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(0, Bytes::from_static(b"a")) @@ -574,30 +589,30 @@ mod test { Chunk::new(6, Bytes::from_static(b"ghi")) ); assert_eq!(x.read(usize::MAX, false), None); - x.insert(8, Bytes::from_static(b"ijkl"), 4); + x.insert(8, Bytes::from_static(b"ijkl"), 4).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(9, Bytes::from_static(b"jkl")) ); assert_eq!(x.read(usize::MAX, false), None); - x.insert(12, Bytes::from_static(b"mno"), 3); + x.insert(12, Bytes::from_static(b"mno"), 3).unwrap(); assert_eq!( next_unordered(&mut x), Chunk::new(12, Bytes::from_static(b"mno")) ); assert_eq!(x.read(usize::MAX, false), None); - x.insert(2, Bytes::from_static(b"cde"), 3); + x.insert(2, Bytes::from_static(b"cde"), 3).unwrap(); assert_eq!(x.read(usize::MAX, false), None); } #[test] fn chunks_dedup() { let mut x = Assembler::new(); - x.insert(3, Bytes::from_static(b"def"), 3); + x.insert(3, Bytes::from_static(b"def"), 3).unwrap(); assert_eq!(x.read(usize::MAX, true), None); - x.insert(0, Bytes::from_static(b"a"), 1); - x.insert(1, Bytes::from_static(b"bcdefghi"), 9); - x.insert(0, Bytes::from_static(b"abcd"), 4); + x.insert(0, Bytes::from_static(b"a"), 1).unwrap(); + x.insert(1, Bytes::from_static(b"bcdefghi"), 9).unwrap(); + x.insert(0, Bytes::from_static(b"abcd"), 4).unwrap(); assert_eq!( x.read(usize::MAX, true), Some(Chunk::new(0, Bytes::from_static(b"abcd"))) @@ -607,34 +622,34 @@ mod test { Some(Chunk::new(4, Bytes::from_static(b"efghi"))) ); assert_eq!(x.read(usize::MAX, true), None); - x.insert(8, Bytes::from_static(b"ijkl"), 4); + x.insert(8, Bytes::from_static(b"ijkl"), 4).unwrap(); assert_eq!( x.read(usize::MAX, true), Some(Chunk::new(9, Bytes::from_static(b"jkl"))) ); assert_eq!(x.read(usize::MAX, true), None); - x.insert(12, Bytes::from_static(b"mno"), 3); + x.insert(12, Bytes::from_static(b"mno"), 3).unwrap(); assert_eq!( x.read(usize::MAX, true), Some(Chunk::new(12, Bytes::from_static(b"mno"))) ); assert_eq!(x.read(usize::MAX, true), None); - x.insert(2, Bytes::from_static(b"cde"), 3); + x.insert(2, Bytes::from_static(b"cde"), 3).unwrap(); assert_eq!(x.read(usize::MAX, true), None); } #[test] fn ordered_eager_discard() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"abc"), 3); + x.insert(0, Bytes::from_static(b"abc"), 3).unwrap(); assert_eq!(x.data.len(), 1); assert_eq!( x.read(usize::MAX, true), Some(Chunk::new(0, Bytes::from_static(b"abc"))) ); - x.insert(0, Bytes::from_static(b"ab"), 2); + x.insert(0, Bytes::from_static(b"ab"), 2).unwrap(); assert_eq!(x.data.len(), 0); - x.insert(2, Bytes::from_static(b"cd"), 2); + x.insert(2, Bytes::from_static(b"cd"), 2).unwrap(); assert_eq!( x.data.peek(), Some(&Buffer::new(3, Bytes::from_static(b"d"), 2)) @@ -644,8 +659,8 @@ mod test { #[test] fn ordered_insert_unordered_read() { let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"abc"), 3); - x.insert(0, Bytes::from_static(b"abc"), 3); + x.insert(0, Bytes::from_static(b"abc"), 3).unwrap(); + x.insert(0, Bytes::from_static(b"abc"), 3).unwrap(); x.ensure_ordering(false).unwrap(); assert_eq!( x.read(3, false), @@ -659,8 +674,8 @@ mod test { // Regression test: bytes read in ordered mode should not be returned again in unordered // mode let mut x = Assembler::new(); - x.insert(0, Bytes::from_static(b"a"), 1); - x.insert(0, Bytes::from_static(b"a"), 1); // duplicate + x.insert(0, Bytes::from_static(b"a"), 1).unwrap(); + x.insert(0, Bytes::from_static(b"a"), 1).unwrap(); // duplicate assert_eq!( x.read(1, true), Some(Chunk::new(0, Bytes::from_static(b"a"))) @@ -779,7 +794,7 @@ mod proptests { match op { Op::Insert { offset, len } => { let bytes = get_slice(&data, offset, len); - asm.insert(offset, bytes, len); + asm.insert(offset, bytes, len).unwrap(); reference.insert(offset, len); } Op::Read { max_len } => { diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 835f022b2f..988258c526 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -4121,7 +4121,9 @@ impl Connection { crypto_space .crypto_stream - .insert(crypto.offset, crypto.data.clone(), payload_len); + .insert(crypto.offset, crypto.data.clone(), payload_len) + .map_err(|_| TransportError::INTERNAL_ERROR("too many gaps in crypto stream buffer"))?; + while let Some(chunk) = crypto_space.crypto_stream.read(usize::MAX, true) { trace!("consumed {} CRYPTO bytes", chunk.bytes.len()); if self.crypto_state.session.read_handshake(&chunk.bytes)? { diff --git a/noq-proto/src/connection/streams/recv.rs b/noq-proto/src/connection/streams/recv.rs index 462adeb563..1209d1daad 100644 --- a/noq-proto/src/connection/streams/recv.rs +++ b/noq-proto/src/connection/streams/recv.rs @@ -79,7 +79,9 @@ impl Recv { // Don't bother storing data or releasing stream-level flow control credit if the stream's // already stopped if !self.stopped { - self.assembler.insert(frame.offset, frame.data, payload_len); + self.assembler + .insert(frame.offset, frame.data, payload_len) + .map_err(|_| TransportError::INTERNAL_ERROR("too many gaps in stream buffer"))?; } Ok((new_bytes, frame.fin && self.stopped)) From dd88e6cf01a805e54a8b566506895a49bdd88715 Mon Sep 17 00:00:00 2001 From: Otto Hermann Date: Wed, 30 Sep 2026 09:40:46 -0700 Subject: [PATCH 2/5] fix(proto): bound assembler chunk count regardless of over-allocation Port of quinn-rs/quinn@4b88c940 (quinn-rs/quinn#2789). The cap above was only checked inside the over-allocation branch, so gapped frames whose allocation matches their length, or frames handled entirely by the unordered dedup loop, still grew the chunk count without bound. Compact once the count passes twice the cap and reject only if compaction cannot bring it back under. Co-Authored-By: Luke Curley Co-Authored-By: Claude Opus 5.5 --- noq-proto/src/connection/assembler.rs | 98 ++++++++++++++++++++++++--- 1 file changed, 90 insertions(+), 8 deletions(-) diff --git a/noq-proto/src/connection/assembler.rs b/noq-proto/src/connection/assembler.rs index 7156deb039..93e4621cf2 100644 --- a/noq-proto/src/connection/assembler.rs +++ b/noq-proto/src/connection/assembler.rs @@ -193,13 +193,13 @@ impl Assembler { } } - if bytes.is_empty() { - return Ok(()); + // No early return when empty: the dedup loop above may already have pushed chunks. + if !bytes.is_empty() { + let buffer = Buffer::new(offset, bytes, allocation_size); + self.buffered += buffer.bytes.len(); + self.allocated += buffer.allocation_size; + self.data.push(buffer); } - let buffer = Buffer::new(offset, bytes, allocation_size); - self.buffered += buffer.bytes.len(); - self.allocated += buffer.allocation_size; - self.data.push(buffer); // `self.buffered` also counts duplicate bytes, therefore we use // `self.end - self.bytes_read` as an upper bound of buffered unique // bytes. This will cause a defragmentation if the amount of duplicate @@ -214,10 +214,11 @@ impl Assembler { // buffered data. The constants are chosen somewhat arbitrarily and try to // balance between defragmentation overhead and over-allocation. let threshold = 32768.max(buffered * 3 / 2); - if over_allocation > threshold { + // Small gapped frames hold over-allocation below the threshold, so bound the count too. + if over_allocation > threshold || self.data.len() > COMPACT_THRESHOLD { self.defragment(); // ngtcp2 uses a threshold of 4000 -- try to be a little more conservative? - if self.data.len() > 1024 { + if self.data.len() > MAX_CHUNKS { return Err(TooManyChunks); } } @@ -356,6 +357,18 @@ pub(crate) struct IllegalOrderedRead; #[derive(Debug)] pub(crate) struct TooManyChunks; +/// Bound on the number of distinct spans kept for a stream +/// +/// Independent of how much memory those spans over-allocate. A frame is rejected only +/// if compaction cannot get the count back down to this. +const MAX_CHUNKS: usize = 1024; + +/// Chunk count past which `insert` compacts before deciding whether to reject +/// +/// Above `MAX_CHUNKS` so a flood of mergeable frames cannot force a defragmentation +/// per frame. +const COMPACT_THRESHOLD: usize = 2 * MAX_CHUNKS; + #[cfg(test)] mod test { use super::*; @@ -684,6 +697,75 @@ mod test { assert_eq!(x.read(1, false), None); // should be None, byte 0 already returned } + #[test] + fn bounded_chunks_under_low_over_allocation() { + // Gapped frames whose `allocation_size` equals their length hold + // `over_allocation` at zero, so that trigger never fires. + let mut x = Assembler::new(); + // Withhold offset 0 so an ordered reader can never drain anything. + let mut offset = 1u64; + let mut result = Ok(()); + for _ in 0..(MAX_CHUNKS * 8) { + result = x.insert(offset, Bytes::from_static(b"gap"), 3); + if result.is_err() { + break; + } + offset += 3 + 1; // 3 data bytes, 1 byte gap + } + assert_matches!(result, Err(TooManyChunks)); + assert!( + x.data.len() <= COMPACT_THRESHOLD + 1, + "chunk count {} exceeded the bound", + x.data.len() + ); + } + + #[test] + fn bounded_chunks_unordered_overlap_flood() { + // Overlapping frames whose tail is already received: the dedup loop pushes + // the fresh head byte and leaves `bytes` empty, and `end` never rises, so + // the flood costs the peer no flow control. + let mut x = Assembler::new(); + x.ensure_ordering(false).unwrap(); + let top = 1_000_000u64; + x.insert(top, Bytes::from_static(b"ab"), 2).unwrap(); + for k in 0..(4 * MAX_CHUNKS as u64) { + x.insert(top - k - 1, Bytes::from_static(b"ab"), 2).unwrap(); + assert!( + x.data.len() <= COMPACT_THRESHOLD + 1, + "chunk count {} exceeded the bound at k={k}", + x.data.len() + ); + } + } + + #[test] + fn bounded_chunks_duplicate_flood() { + // Duplicates against a stream already at the cap. Ordered mode does not dedup, + // so each one pushes a chunk; they must not be rejected, or compact every frame. + let mut x = Assembler::new(); + // Withhold offset 0 so nothing can be drained. + for i in 0..MAX_CHUNKS as u64 { + x.insert(1 + i * 4, Bytes::from_static(b"abc"), 3).unwrap(); + } + let mut max_len = x.data.len(); + for _ in 0..(3 * MAX_CHUNKS) { + x.insert(1, Bytes::from_static(b"abc"), 3) + .expect("duplicate flood must not be rejected"); + max_len = max_len.max(x.data.len()); + assert!( + x.data.len() <= COMPACT_THRESHOLD + 1, + "chunk count {} exceeded the bound", + x.data.len() + ); + } + // Compacting on every frame would pin the count at `MAX_CHUNKS`. + assert!( + max_len > MAX_CHUNKS, + "buffer compacted on every frame (max observed len {max_len})" + ); + } + fn next_unordered(x: &mut Assembler) -> Chunk { x.read(usize::MAX, false).unwrap() } From c2eb72767beb16ca9b89cc8381c9fa12cf66d938 Mon Sep 17 00:00:00 2001 From: Eanna Lyons Date: Wed, 30 Sep 2026 09:45:20 -0700 Subject: [PATCH 3/5] fix(proto): coalesce contiguous chunks during defragment Port of quinn-rs/quinn@c3f8b098 (quinn-rs/quinn#2814). defragment now combines contiguous chunks smaller than buffered / MAX_CHUNKS, so a large stream_receive_window with a slow in-order reader no longer trips TooManyChunks on lossless data. Chunks above the minimum size are left alone so they are not copied on every compaction. Co-Authored-By: Luke Curley Co-Authored-By: Claude Opus 5.5 --- noq-proto/src/connection/assembler.rs | 54 +++++++++++++++++++++++---- 1 file changed, 47 insertions(+), 7 deletions(-) diff --git a/noq-proto/src/connection/assembler.rs b/noq-proto/src/connection/assembler.rs index 93e4621cf2..c41aae6c5c 100644 --- a/noq-proto/src/connection/assembler.rs +++ b/noq-proto/src/connection/assembler.rs @@ -1,5 +1,5 @@ use std::{ - cmp::Ordering, + cmp::{Ordering, max}, collections::{BinaryHeap, binary_heap::PeekMut}, mem, }; @@ -107,6 +107,8 @@ impl Assembler { /// This makes sure we're not unnecessarily holding on to many larger allocations. /// We merge contiguous chunks in the process of doing so. fn defragment(&mut self) { + // Chunks smaller than min_chunk_size are merged regardless of fragmentation + let min_chunk_size = max(self.buffered.div_ceil(MAX_CHUNKS), MIN_RETAINED_CHUNK_SIZE); let new = BinaryHeap::with_capacity(self.data.len()); let old = mem::replace(&mut self.data, new); let mut buffers = old.into_sorted_vec(); @@ -118,7 +120,7 @@ impl Assembler { let size = chunk.bytes.len(); offset = chunk.offset + size as u64; self.buffered += size; - if !chunk.defragmented { + if !chunk.defragmented || size < min_chunk_size { fragmented_buffered += size; } } @@ -126,11 +128,12 @@ impl Assembler { let mut buffer = BytesMut::with_capacity(fragmented_buffered); let mut offset = self.bytes_read; for chunk in buffers.into_iter().rev() { - if chunk.defragmented { - // bytes might be empty after try_mark_defragment - if !chunk.bytes.is_empty() { - self.data.push(chunk); - } + // bytes might be empty after try_mark_defragment + if chunk.bytes.is_empty() { + continue; + } + if chunk.defragmented && chunk.bytes.len() >= min_chunk_size { + self.data.push(chunk); continue; } // Overlap is resolved by try_mark_defragment @@ -363,6 +366,12 @@ pub(crate) struct TooManyChunks; /// if compaction cannot get the count back down to this. const MAX_CHUNKS: usize = 1024; +/// Minimum size of a defragmented chunk that is retained without coalescing. +/// +/// Chunks below this size will be copied on every compaction. 128 bytes balances +/// being cheap to copy while reducing the overhead of many small buffer nodes. +const MIN_RETAINED_CHUNK_SIZE: usize = 128; + /// Chunk count past which `insert` compacts before deciding whether to reject /// /// Above `MAX_CHUNKS` so a flood of mergeable frames cannot force a defragmentation @@ -697,6 +706,37 @@ mod test { assert_eq!(x.read(1, false), None); // should be None, byte 0 already returned } + #[test] + fn ordered_lossless_stream_is_not_rejected() { + const FRAME_LEN: usize = 32; + + let mut x = Assembler::new(); + let data = Bytes::from(vec![0; FRAME_LEN]); + let mut offset = 0u64; + while offset < (2 * COMPACT_THRESHOLD * FRAME_LEN) as u64 { + x.insert(offset, data.clone(), FRAME_LEN) + .expect("contiguous lossless data must not be rejected"); + offset += FRAME_LEN as u64; + } + } + + #[test] + fn defrag_respects_max_chunks_at_rounding_boundary() { + let mut x = Assembler::new(); + let data = Bytes::from(vec![0; MIN_RETAINED_CHUNK_SIZE]); + let mut offset = 0; + // Fill with minimium sized chunks. + for _ in 0..MAX_CHUNKS { + x.insert(offset, data.clone(), data.len()).unwrap(); + offset += data.len() as u64; + } + // Add 1 extra chunk, defragment must coalesce the previous + // chunks to stay within MAX_CHUNKS + x.insert(offset, Bytes::from_static(b"x"), 1).unwrap(); + x.defragment(); + assert!(x.data.len() <= MAX_CHUNKS); + } + #[test] fn bounded_chunks_under_low_over_allocation() { // Gapped frames whose `allocation_size` equals their length hold From 19e7bc676a0363482a7557065e6b2800121db505 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 28 Sep 2026 17:23:45 -0700 Subject: [PATCH 4/5] chore: release 2.0.1 Bumps the shared workspace version and the internal pins, and records #25 (reassembly cap, RUSTSEC-2026-0185), #19, and #23 in CHANGELOG-MOQ.md. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG-MOQ.md | 10 ++++++++++ Cargo.lock | 8 ++++---- Cargo.toml | 2 +- noq/Cargo.toml | 4 ++-- web-transport-moq/Cargo.toml | 2 +- 5 files changed, 18 insertions(+), 8 deletions(-) diff --git a/CHANGELOG-MOQ.md b/CHANGELOG-MOQ.md index f97f455b8c..d7773148af 100644 --- a/CHANGELOG-MOQ.md +++ b/CHANGELOG-MOQ.md @@ -6,6 +6,16 @@ covers the parent. Each entry names the parent commit it carries and every change the fork carries on top, with its upstream status, so an advisory against the parent can be checked against a release. +## 2.0.1 + +Parent: n0-computer/noq [`1a26a8b0`](https://github.com/n0-computer/noq/commit/1a26a8b064d21e316fe6769f068617975bd8a27b), unchanged since 1.3.0. + +- [#25](https://github.com/moq-dev/noq/pull/25) bound the stream and CRYPTO reassembly buffers at 1024 chunks after compaction, closing the connection with `INTERNAL_ERROR` past it ([RUSTSEC-2026-0185](https://rustsec.org/advisories/RUSTSEC-2026-0185.html)). Carries [quinn-rs/quinn#2694](https://github.com/quinn-rs/quinn/pull/2694), [#2789](https://github.com/quinn-rs/quinn/pull/2789), and [#2814](https://github.com/quinn-rs/quinn/pull/2814), matching quinn-proto 0.11.18. Also in 1.3.3 ([#26](https://github.com/moq-dev/noq/pull/26)). Upstream: [n0-computer/noq#828](https://github.com/n0-computer/noq/pull/828) (open). +- [#19](https://github.com/moq-dev/noq/pull/19) report a stream reset before its WebTransport header as the reset (`WebTransportError::ReadError`), logged at debug, instead of `UnknownSession` at WARN. `UnknownSession` now means only a session ID mismatch. Also released as 1.3.3 from `release/1.3` ([#20](https://github.com/moq-dev/noq/pull/20)). Fork-only. +- [#23](https://github.com/moq-dev/noq/pull/23) keep a `web-transport-moq` session's HTTP/3 control and QPACK streams open until its close capsule is delivered, so browsers see the close code and reason instead of "Connection lost.". Also in 1.3.3 ([#24](https://github.com/moq-dev/noq/pull/24)). Fork-only. + +No API change. A peer that exceeds the reassembly cap is now closed. + ## 2.0.0 Parent: n0-computer/noq [`1a26a8b0`](https://github.com/n0-computer/noq/commit/1a26a8b064d21e316fe6769f068617975bd8a27b), unchanged since 1.3.0. diff --git a/Cargo.lock b/Cargo.lock index b5a05714b5..94c343e6cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1876,7 +1876,7 @@ dependencies = [ [[package]] name = "moq-noq" -version = "2.0.0" +version = "2.0.1" dependencies = [ "anyhow", "async-executor", @@ -1919,7 +1919,7 @@ dependencies = [ [[package]] name = "moq-noq-proto" -version = "2.0.0" +version = "2.0.1" dependencies = [ "aes-gcm", "arbitrary", @@ -1959,7 +1959,7 @@ dependencies = [ [[package]] name = "moq-noq-udp" -version = "2.0.0" +version = "2.0.1" dependencies = [ "cfg_aliases", "criterion", @@ -3531,7 +3531,7 @@ dependencies = [ [[package]] name = "web-transport-moq" -version = "2.0.0" +version = "2.0.1" dependencies = [ "anyhow", "bytes", diff --git a/Cargo.toml b/Cargo.toml index f775102cb1..3d0401fed0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ default-members = ["noq", "noq-proto", "noq-udp", "web-transport-moq", "bench", resolver = "2" [workspace.package] -version = "2.0.0" +version = "2.0.1" rust-version = "1.88" edition = "2024" license = "MIT OR Apache-2.0" diff --git a/noq/Cargo.toml b/noq/Cargo.toml index ff3b0af2b1..5b91ff11a3 100644 --- a/noq/Cargo.toml +++ b/noq/Cargo.toml @@ -67,13 +67,13 @@ derive_more = { workspace = true } futures-io = { workspace = true, optional = true } rustc-hash = { workspace = true } pin-project-lite = { workspace = true } -proto = { package = "moq-noq-proto", path = "../noq-proto", version = "2.0.0", default-features = false } +proto = { package = "moq-noq-proto", path = "../noq-proto", version = "2.0.1", default-features = false } rustls = { workspace = true, optional = true } smol = { workspace = true, optional = true } thiserror = { workspace = true } tracing = { workspace = true } tokio = { workspace = true } -udp = { package = "moq-noq-udp", path = "../noq-udp", version = "2.0.0", default-features = false, features = ["tracing"] } +udp = { package = "moq-noq-udp", path = "../noq-udp", version = "2.0.1", default-features = false, features = ["tracing"] } # Fix minimal dependencies for indirect deps async-global-executor = { workspace = true, optional = true } diff --git a/web-transport-moq/Cargo.toml b/web-transport-moq/Cargo.toml index 4143fb5565..ada0b5cb97 100644 --- a/web-transport-moq/Cargo.toml +++ b/web-transport-moq/Cargo.toml @@ -28,7 +28,7 @@ bytes = "1" futures = "0.3" http = "1" kio = "0.5.4" -noq = { package = "moq-noq", path = "../noq", version = "2.0.0", default-features = false, features = [ +noq = { package = "moq-noq", path = "../noq", version = "2.0.1", default-features = false, features = [ "tracing-log", "platform-verifier", "runtime-tokio", From b29f992a91adbe7254135e27e2ab94111ccd2842 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 17:20:47 -0700 Subject: [PATCH 5/5] chore: cover #27 and the unordered compaction fix in 2.0.1 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG-MOQ.md | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/CHANGELOG-MOQ.md b/CHANGELOG-MOQ.md index d7773148af..873aa8e062 100644 --- a/CHANGELOG-MOQ.md +++ b/CHANGELOG-MOQ.md @@ -10,11 +10,15 @@ against the parent can be checked against a release. Parent: n0-computer/noq [`1a26a8b0`](https://github.com/n0-computer/noq/commit/1a26a8b064d21e316fe6769f068617975bd8a27b), unchanged since 1.3.0. -- [#25](https://github.com/moq-dev/noq/pull/25) bound the stream and CRYPTO reassembly buffers at 1024 chunks after compaction, closing the connection with `INTERNAL_ERROR` past it ([RUSTSEC-2026-0185](https://rustsec.org/advisories/RUSTSEC-2026-0185.html)). Carries [quinn-rs/quinn#2694](https://github.com/quinn-rs/quinn/pull/2694), [#2789](https://github.com/quinn-rs/quinn/pull/2789), and [#2814](https://github.com/quinn-rs/quinn/pull/2814), matching quinn-proto 0.11.18. Also in 1.3.3 ([#26](https://github.com/moq-dev/noq/pull/26)). Upstream: [n0-computer/noq#828](https://github.com/n0-computer/noq/pull/828) (open). -- [#19](https://github.com/moq-dev/noq/pull/19) report a stream reset before its WebTransport header as the reset (`WebTransportError::ReadError`), logged at debug, instead of `UnknownSession` at WARN. `UnknownSession` now means only a session ID mismatch. Also released as 1.3.3 from `release/1.3` ([#20](https://github.com/moq-dev/noq/pull/20)). Fork-only. +- [#25](https://github.com/moq-dev/noq/pull/25) bound the stream and CRYPTO reassembly buffers at 1024 chunks after compaction, closing the connection with `INTERNAL_ERROR` past it ([RUSTSEC-2026-0185](https://rustsec.org/advisories/RUSTSEC-2026-0185.html)). Carries [quinn-rs/quinn#2694](https://github.com/quinn-rs/quinn/pull/2694), [#2789](https://github.com/quinn-rs/quinn/pull/2789), and [#2814](https://github.com/quinn-rs/quinn/pull/2814), matching quinn-proto 0.11.18, plus a fix so compacting an unordered stream keeps its unread data (upstream noq starts compaction at `bytes_read`, which the new count trigger made reachable). Also in 1.3.3 ([#26](https://github.com/moq-dev/noq/pull/26)). Upstream: [n0-computer/noq#828](https://github.com/n0-computer/noq/pull/828) (open). +- [#19](https://github.com/moq-dev/noq/pull/19) report a stream reset before its WebTransport header as the reset (`WebTransportError::ReadError`), logged at debug, instead of `UnknownSession` at WARN. `UnknownSession` now means only a missing or mismatched session ID. Also in 1.3.3 ([#20](https://github.com/moq-dev/noq/pull/20)). Fork-only. - [#23](https://github.com/moq-dev/noq/pull/23) keep a `web-transport-moq` session's HTTP/3 control and QPACK streams open until its close capsule is delivered, so browsers see the close code and reason instead of "Connection lost.". Also in 1.3.3 ([#24](https://github.com/moq-dev/noq/pull/24)). Fork-only. +- [#27](https://github.com/moq-dev/noq/pull/27) send a raw QUIC session's (`Session::raw`) stream reset and stop codes as is, instead of mapping them into the HTTP/3 WebTransport range, and read a peer's code as is. Not in 1.3.3. Fork-only. -No API change. A peer that exceeds the reassembly cap is now closed. +No API change. Wire: + +- A peer that exceeds the reassembly cap is now closed. +- Raw QUIC stream codes are compatible in one direction only: 2.0.1 still reads an older raw peer's mapped codes, but an older raw peer reads 2.0.1's codes as `InvalidReset` / `InvalidStopped`. HTTP/3 sessions are unchanged. ## 2.0.0