From 5fb9182cde12f5cb2caada1282ee19b94f5da3c1 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 28 Aug 2026 09:51:31 +0200 Subject: [PATCH 1/5] Add a regression test showing that a read after a dropped read_to_end panics. --- noq/src/tests.rs | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/noq/src/tests.rs b/noq/src/tests.rs index adcd75682e..9f1c86f988 100755 --- a/noq/src/tests.rs +++ b/noq/src/tests.rs @@ -1730,6 +1730,41 @@ async fn recv_stream_cancel_stop_drop() { ); } +/// Dropping a pending `read_to_end` future must not make a subsequent ordered read panic. +#[tokio::test] +async fn recv_stream_cancel_read_to_end_then_ordered_read() { + let _guard = subscribe(); + let factory = EndpointFactory::new(); + let server = factory.endpoint("server"); + let server_addr = server.local_addr().unwrap(); + let client = factory.endpoint("client"); + + tokio::join!( + async { + let conn = server.accept().await.unwrap().await.unwrap(); + let mut recv = conn.accept_uni().await.unwrap(); + { + let fut = pin!(recv.read_to_end(usize::MAX)); + let mut cx = Context::from_waker(Waker::noop()); + assert!(fut.poll(&mut cx).is_pending()); + } + + let mut buf = [0; 1]; + recv.read(&mut buf).await.unwrap(); + }, + async { + let conn = client + .connect(server_addr, "localhost") + .unwrap() + .await + .unwrap(); + let mut send = conn.open_uni().await.unwrap(); + send.write_all(b"hello").await.unwrap(); + std::future::pending::<()>().await; + }, + ); +} + /// Regression test for an `active_connections` underflow panic in the endpoint driver. /// /// `ConnectionSet::insert` used to only increment `active_connections` when the endpoint From 1be34d6ef47228452e08acb89f906d04ba4bc52a Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 28 Aug 2026 10:00:23 +0200 Subject: [PATCH 2/5] fix: change read_to_end to use the ordered API This fixes the issue that you can get into a state where you drop read_to_end and then try an ordered read, which will fail with the current API. TBH allowing read to end for a stream that already had unordered reads and then have "gaps containing arbitrary data" is weird anyway. And the performance benefit of unordered reads in typical read to end use cases is very modest. --- noq/src/recv_stream.rs | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/noq/src/recv_stream.rs b/noq/src/recv_stream.rs index a5ba953918..9a3b379217 100644 --- a/noq/src/recv_stream.rs +++ b/noq/src/recv_stream.rs @@ -249,11 +249,7 @@ impl RecvStream { /// Convenience method to read all remaining data into a buffer /// /// Fails with [`ReadToEndError::TooLong`] on reading more than `size_limit` bytes, discarding - /// all data read. Uses unordered reads to be more efficient than using `AsyncRead` would - /// allow. `size_limit` should be set to limit worst-case memory use. - /// - /// If unordered reads have already been made, the resulting buffer may have gaps containing - /// arbitrary data. + /// all data read. `size_limit` should be set to limit worst-case memory use. /// /// This operation is *not* cancel-safe. /// @@ -540,7 +536,7 @@ impl Future for ReadToEnd<'_> { type Output = Result, ReadToEndError>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { loop { - match ready!(self.stream.poll_read_chunk(cx, usize::MAX, false))? { + match ready!(self.stream.poll_read_chunk(cx, usize::MAX, true))? { Some(chunk) => { self.start = self.start.min(chunk.offset); let end = chunk.bytes.len() as u64 + chunk.offset; From ac07b31167076788b503a351fcab885ab01c5818 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 28 Aug 2026 10:10:11 +0200 Subject: [PATCH 3/5] fix the test now that it doesn't panic anymore --- noq/src/tests.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/noq/src/tests.rs b/noq/src/tests.rs index 9f1c86f988..ade0be7b2a 100755 --- a/noq/src/tests.rs +++ b/noq/src/tests.rs @@ -1738,6 +1738,7 @@ async fn recv_stream_cancel_read_to_end_then_ordered_read() { let server = factory.endpoint("server"); let server_addr = server.local_addr().unwrap(); let client = factory.endpoint("client"); + let ordered_read_done = tokio::sync::SetOnce::new(); tokio::join!( async { @@ -1750,7 +1751,10 @@ async fn recv_stream_cancel_read_to_end_then_ordered_read() { } let mut buf = [0; 1]; - recv.read(&mut buf).await.unwrap(); + let fut = pin!(recv.read(&mut buf)); + let mut cx = Context::from_waker(Waker::noop()); + assert!(fut.poll(&mut cx).is_pending()); + ordered_read_done.set(()).unwrap(); }, async { let conn = client @@ -1760,7 +1764,7 @@ async fn recv_stream_cancel_read_to_end_then_ordered_read() { .unwrap(); let mut send = conn.open_uni().await.unwrap(); send.write_all(b"hello").await.unwrap(); - std::future::pending::<()>().await; + ordered_read_done.wait().await; }, ); } From 2bd9cdcbdfb777671b0b215bd46ffa3e0e0910ba Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 28 Aug 2026 11:06:31 +0200 Subject: [PATCH 4/5] Revert read_to_end to use unordered API we call the ordered api once to trigger a ClosedStream if we are already in unordered mode. Also update the test. --- noq/src/recv_stream.rs | 21 ++++++++++++--------- noq/src/tests.rs | 22 ++++++++++++++++------ 2 files changed, 28 insertions(+), 15 deletions(-) diff --git a/noq/src/recv_stream.rs b/noq/src/recv_stream.rs index 9a3b379217..fa5386090f 100644 --- a/noq/src/recv_stream.rs +++ b/noq/src/recv_stream.rs @@ -249,9 +249,11 @@ impl RecvStream { /// Convenience method to read all remaining data into a buffer /// /// Fails with [`ReadToEndError::TooLong`] on reading more than `size_limit` bytes, discarding - /// all data read. `size_limit` should be set to limit worst-case memory use. + /// all data read. Uses unordered reads to be more efficient than using `AsyncRead` would + /// allow. `size_limit` should be set to limit worst-case memory use. /// - /// This operation is *not* cancel-safe. + /// This operation is *not* cancel-safe. If cancelled after it has begun reading, further read + /// operations on the stream return [`ReadError::ClosedStream`]. /// /// [`ReadToEndError::TooLong`]: crate::ReadToEndError::TooLong pub async fn read_to_end(&mut self, size_limit: usize) -> Result, ReadToEndError> { @@ -261,6 +263,7 @@ impl RecvStream { read: Vec::new(), start: u64::MAX, end: 0, + first_read: true, } .await } @@ -387,12 +390,7 @@ impl RecvStream { let mut recv = conn.inner.recv_stream(self.stream); let mut chunks = recv.read(ordered).map_err(|e| match e { ReadableError::ClosedStream => ReadError::ClosedStream, - ReadableError::IllegalOrderedRead => { - // We should never get here because the only way to do unordered reads is - // via UnorderedRecvStream, which allows only unordered reads. It is not - // possible to get a RecvStream from an UnorderedRecvStream. - unreachable!("ordered read after unordered read") - } + ReadableError::IllegalOrderedRead => ReadError::ClosedStream, })?; let status = read_fn(&mut chunks); if chunks.finalize().should_transmit() { @@ -530,13 +528,18 @@ struct ReadToEnd<'a> { start: u64, end: u64, size_limit: usize, + first_read: bool, } impl Future for ReadToEnd<'_> { type Output = Result, ReadToEndError>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { loop { - match ready!(self.stream.poll_read_chunk(cx, usize::MAX, true))? { + // the first call to poll_read_chunk is ordered to trigger a ClosedStream + // error if the underlying stream is already in unordered mode. + let ordered = self.first_read; + self.first_read = false; + match ready!(self.stream.poll_read_chunk(cx, usize::MAX, ordered))? { Some(chunk) => { self.start = self.start.min(chunk.offset); let end = chunk.bytes.len() as u64 + chunk.offset; diff --git a/noq/src/tests.rs b/noq/src/tests.rs index ade0be7b2a..f4bc8e9afa 100755 --- a/noq/src/tests.rs +++ b/noq/src/tests.rs @@ -1730,9 +1730,10 @@ async fn recv_stream_cancel_stop_drop() { ); } -/// Dropping a pending `read_to_end` future must not make a subsequent ordered read panic. +/// Dropping a pending `read_to_end` future makes subsequent reads fail because `read_to_end` uses +/// the unordered API internally. #[tokio::test] -async fn recv_stream_cancel_read_to_end_then_ordered_read() { +async fn recv_stream_cancel_read_to_end_then_ordered_read_is_closed() { let _guard = subscribe(); let factory = EndpointFactory::new(); let server = factory.endpoint("server"); @@ -1750,10 +1751,19 @@ async fn recv_stream_cancel_read_to_end_then_ordered_read() { assert!(fut.poll(&mut cx).is_pending()); } - let mut buf = [0; 1]; - let fut = pin!(recv.read(&mut buf)); - let mut cx = Context::from_waker(Waker::noop()); - assert!(fut.poll(&mut cx).is_pending()); + { + let mut buf = [0; 1]; + let fut = pin!(recv.read(&mut buf)); + let mut cx = Context::from_waker(Waker::noop()); + assert!(matches!( + fut.poll(&mut cx), + Poll::Ready(Err(crate::ReadError::ClosedStream)) + )); + } + assert_eq!( + recv.read_to_end(usize::MAX).await, + Err(crate::ReadToEndError::Read(crate::ReadError::ClosedStream)) + ); ordered_read_done.set(()).unwrap(); }, async { From 0ee8794f7af9195f459d8b0ae6066eda535cd388 Mon Sep 17 00:00:00 2001 From: Ruediger Klaehn Date: Fri, 28 Aug 2026 12:19:38 +0200 Subject: [PATCH 5/5] Add is_ordered fn to noq-proto and use it from read_to_end before creating the future The future now can remain completely unchanged. --- noq-proto/src/connection/assembler.rs | 4 ++++ noq-proto/src/connection/streams/mod.rs | 19 ++++++++++++++- noq-proto/src/connection/streams/state.rs | 28 +++++++++++++++++++++++ noq/src/recv_stream.rs | 22 ++++++++++++------ 4 files changed, 65 insertions(+), 8 deletions(-) diff --git a/noq-proto/src/connection/assembler.rs b/noq-proto/src/connection/assembler.rs index 41dd015672..ca0a8d9a06 100644 --- a/noq-proto/src/connection/assembler.rs +++ b/noq-proto/src/connection/assembler.rs @@ -37,6 +37,10 @@ impl Assembler { self.data.clear(); } + pub(super) fn is_ordered(&self) -> bool { + self.state.is_ordered() + } + pub(super) fn ensure_ordering(&mut self, ordered: bool) -> Result<(), IllegalOrderedRead> { if ordered && !self.state.is_ordered() { return Err(IllegalOrderedRead); diff --git a/noq-proto/src/connection/streams/mod.rs b/noq-proto/src/connection/streams/mod.rs index 1b4bcc4dc7..56fd178e45 100644 --- a/noq-proto/src/connection/streams/mod.rs +++ b/noq-proto/src/connection/streams/mod.rs @@ -10,7 +10,7 @@ use tracing::trace; use super::spaces::Retransmits; use crate::{ Dir, StreamId, VarInt, - connection::streams::state::{get_or_insert_recv, get_or_insert_send}, + connection::streams::state::{StreamRecv, get_or_insert_recv, get_or_insert_send}, frame, }; @@ -110,6 +110,23 @@ pub struct RecvStream<'a> { } impl RecvStream<'_> { + /// Whether this stream is still in ordered read mode. + /// + /// A stream switches permanently to unordered mode when [`Self::read`] is called with + /// `ordered` set to `false`. + pub fn is_ordered(&self) -> Result { + let Some(stream) = self.state.recv.get(&self.id) else { + return Err(ClosedStream { _private: () }); + }; + let Some(stream) = stream.as_ref().and_then(StreamRecv::as_open_recv) else { + return Ok(true); + }; + if stream.stopped { + return Err(ClosedStream { _private: () }); + } + Ok(stream.assembler.is_ordered()) + } + /// Read from the given recv stream /// /// `max_length` limits the maximum size of the returned `Bytes` value; passing `usize::MAX` diff --git a/noq-proto/src/connection/streams/state.rs b/noq-proto/src/connection/streams/state.rs index 6eadee34f0..736a350e33 100644 --- a/noq-proto/src/connection/streams/state.rs +++ b/noq-proto/src/connection/streams/state.rs @@ -1194,6 +1194,7 @@ mod tests { assert!(recv.stop(0u32.into()).is_err()); assert_eq!(recv.read(true).err(), Some(ReadableError::ClosedStream)); assert_eq!(recv.read(false).err(), Some(ReadableError::ClosedStream)); + assert!(recv.is_ordered().is_err()); assert_eq!(client.local_max_data - initial_max, 32); assert_eq!( @@ -1214,6 +1215,33 @@ mod tests { assert!(!client.recv.contains_key(&id)); } + #[test] + fn recv_stream_ordering_mode() { + let mut client = make(Side::Client); + let id = StreamId::new(Side::Server, Dir::Uni, 0); + let _ = client + .received( + frame::Stream { + id, + offset: 0, + fin: false, + data: Bytes::from_static(b"hello"), + }, + 5, + ) + .unwrap(); + + let mut pending = Retransmits::default(); + let mut recv = RecvStream { + id, + state: &mut client, + pending: &mut pending, + }; + assert_eq!(recv.is_ordered(), Ok(true)); + let _ = recv.read(false).unwrap().finalize(); + assert_eq!(recv.is_ordered(), Ok(false)); + } + #[test] fn stopped_reset() { let mut client = make(Side::Client); diff --git a/noq/src/recv_stream.rs b/noq/src/recv_stream.rs index fa5386090f..26c1c878ab 100644 --- a/noq/src/recv_stream.rs +++ b/noq/src/recv_stream.rs @@ -200,6 +200,17 @@ impl RecvStream { }) } + fn is_ordered(&self) -> Result { + let mut conn = self.conn.lock_without_waking("RecvStream::is_ordered"); + if self.is_0rtt { + conn.check_0rtt().map_err(|()| ReadError::ZeroRttRejected)?; + } + conn.inner + .recv_stream(self.stream) + .is_ordered() + .map_err(|_| ReadError::ClosedStream) + } + /// Reads the next segments of data. /// /// Fills `bufs` with the segments of data beginning immediately after the last data yielded @@ -257,13 +268,15 @@ impl RecvStream { /// /// [`ReadToEndError::TooLong`]: crate::ReadToEndError::TooLong pub async fn read_to_end(&mut self, size_limit: usize) -> Result, ReadToEndError> { + if !self.is_ordered()? { + return Err(ReadError::ClosedStream.into()); + } ReadToEnd { stream: self, size_limit, read: Vec::new(), start: u64::MAX, end: 0, - first_read: true, } .await } @@ -528,18 +541,13 @@ struct ReadToEnd<'a> { start: u64, end: u64, size_limit: usize, - first_read: bool, } impl Future for ReadToEnd<'_> { type Output = Result, ReadToEndError>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { loop { - // the first call to poll_read_chunk is ordered to trigger a ClosedStream - // error if the underlying stream is already in unordered mode. - let ordered = self.first_read; - self.first_read = false; - match ready!(self.stream.poll_read_chunk(cx, usize::MAX, ordered))? { + match ready!(self.stream.poll_read_chunk(cx, usize::MAX, false))? { Some(chunk) => { self.start = self.start.min(chunk.offset); let end = chunk.bytes.len() as u64 + chunk.offset;