From b7a4d3663e904d5d9b20fa3d2fa28295aacfb4f5 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Wed, 10 Jun 2026 18:29:37 +0200 Subject: [PATCH 1/5] feat: reset_stream_at TransportParameter --- noq-proto/src/config/mod.rs | 23 +++++++++++------------ noq-proto/src/frame.rs | 2 ++ noq-proto/src/transport_parameters.rs | 26 +++++++++++++++++++++++++- 3 files changed, 38 insertions(+), 13 deletions(-) diff --git a/noq-proto/src/config/mod.rs b/noq-proto/src/config/mod.rs index 32b944223e..ffacae1d31 100644 --- a/noq-proto/src/config/mod.rs +++ b/noq-proto/src/config/mod.rs @@ -36,13 +36,15 @@ pub use transport::{AckFrequencyConfig, IdleTimeout, MtuDiscoveryConfig, Transpo /// Global configuration for the endpoint, affecting all connections /// /// Default values should be suitable for most internet applications. -#[derive(Clone)] +#[derive(Clone, derive_more::Debug)] pub struct EndpointConfig { + #[debug("HmacKey")] pub(crate) reset_key: Arc, pub(crate) max_udp_payload_size: VarInt, /// CID generator factory /// /// Create a cid generator for local cid in Endpoint struct + #[debug("ConnectionIdGenerator")] pub(crate) connection_id_generator_factory: Arc Box + Send + Sync>, pub(crate) supported_versions: Vec, @@ -51,6 +53,7 @@ pub struct EndpointConfig { pub(crate) min_reset_interval: Duration, /// Optional seed to be used internally for random number generation pub(crate) rng_seed: Option<[u8; 32]>, + pub(crate) reset_stream_at: bool, } impl EndpointConfig { @@ -66,6 +69,7 @@ impl EndpointConfig { grease_quic_bit: true, min_reset_interval: Duration::from_millis(20), rng_seed: None, + reset_stream_at: true, } } @@ -162,18 +166,13 @@ impl EndpointConfig { self.rng_seed = seed; self } -} -impl fmt::Debug for EndpointConfig { - fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { - fmt.debug_struct("EndpointConfig") - // reset_key not debug - .field("max_udp_payload_size", &self.max_udp_payload_size) - // cid_generator_factory not debug - .field("supported_versions", &self.supported_versions) - .field("grease_quic_bit", &self.grease_quic_bit) - .field("rng_seed", &self.rng_seed) - .finish_non_exhaustive() + /// Enables the QUIC Stream Resets with Partial Delivery draft-07 extentsion. + /// + /// + pub fn reliable_stream_reset(&mut self, value: bool) -> &mut Self { + self.reset_stream_at = value; + self } } diff --git a/noq-proto/src/frame.rs b/noq-proto/src/frame.rs index 51ebd3befa..b7fa6c0201 100644 --- a/noq-proto/src/frame.rs +++ b/noq-proto/src/frame.rs @@ -133,6 +133,8 @@ pub enum FrameType { ReachOutAtIpv6, #[assoc(to_u64 = 0x3d7f94)] RemoveAddress, + // #[assoc(to_u64 = 0x24)] + // ResetStreamAt, } /// Encounter a frame ID that was not valid. diff --git a/noq-proto/src/transport_parameters.rs b/noq-proto/src/transport_parameters.rs index 7dc2907c99..b721c2071c 100644 --- a/noq-proto/src/transport_parameters.rs +++ b/noq-proto/src/transport_parameters.rs @@ -122,6 +122,9 @@ macro_rules! make_struct { /// Nat traversal draft pub max_remote_nat_traversal_addresses: Option, + + /// QUIC Stream Resets with Partial Delivery + pub(crate) reset_stream_at: bool, } // We deliberately don't implement the `Default` trait, since that would be public, and @@ -148,6 +151,7 @@ macro_rules! make_struct { address_discovery_role: address_discovery::Role::default(), initial_max_path_id: None, max_remote_nat_traversal_addresses: None, + reset_stream_at: false, } } } @@ -198,6 +202,7 @@ impl TransportParameters { address_discovery_role: config.address_discovery_role, initial_max_path_id: config.get_initial_max_path_id(), max_remote_nat_traversal_addresses: config.max_remote_nat_traversal_addresses, + reset_stream_at: endpoint_config.reset_stream_at, ..Self::default() } } @@ -216,6 +221,7 @@ impl TransportParameters { || cached.grease_quic_bit && !self.grease_quic_bit || cached.address_discovery_role != self.address_discovery_role || cached.max_remote_nat_traversal_addresses != self.max_remote_nat_traversal_addresses + || cached.reset_stream_at != self.reset_stream_at { return Err(TransportError::PROTOCOL_VIOLATION( "0-RTT accepted with incompatible transport parameters", @@ -423,6 +429,12 @@ impl TransportParameters { w.write(val.get()); } } + TransportParameterId::ResetStreamAt => { + if self.reset_stream_at { + w.write_var(id as u64); + w.write_var(0); + } + } id => { macro_rules! write_params { {$($(#[$doc:meta])* $name:ident ($id:ident) = $default:expr,)*} => { @@ -559,6 +571,12 @@ impl TransportParameters { params.max_remote_nat_traversal_addresses = Some(value); } + TransportParameterId::ResetStreamAt => { + if len != 0 || params.reset_stream_at { + return Err(Error::Malformed); + } + params.reset_stream_at = true; + } _ => { macro_rules! parse { {$($(#[$doc:meta])* $name:ident ($id:ident) = $default:expr,)*} => { @@ -731,11 +749,14 @@ pub(crate) enum TransportParameterId { // inspired by https://www.ietf.org/archive/id/draft-seemann-quic-nat-traversal-02.html, // simplified to n0's own protocol. N0NatTraversal = 0x3d7f91120401, + + // https://datatracker.ietf.org/doc/html/draft-ietf-quic-reliable-stream-reset + ResetStreamAt = 0x17f7586d2cb571, } impl TransportParameterId { /// Array with all supported transport parameter IDs - const SUPPORTED: [Self; 24] = [ + const SUPPORTED: [Self; 25] = [ Self::MaxIdleTimeout, Self::MaxUdpPayloadSize, Self::InitialMaxData, @@ -760,6 +781,7 @@ impl TransportParameterId { Self::ObservedAddr, Self::InitialMaxPathId, Self::N0NatTraversal, + Self::ResetStreamAt, ]; } @@ -802,6 +824,7 @@ impl TryFrom for TransportParameterId { id if Self::ObservedAddr == id => Self::ObservedAddr, id if Self::InitialMaxPathId == id => Self::InitialMaxPathId, id if Self::N0NatTraversal == id => Self::N0NatTraversal, + id if Self::ResetStreamAt == id => Self::ResetStreamAt, _ => return Err(()), }; Ok(param) @@ -843,6 +866,7 @@ mod test { address_discovery_role: address_discovery::Role::send_only(), initial_max_path_id: Some(PathId::MAX), max_remote_nat_traversal_addresses: Some(5u8.try_into().unwrap()), + reset_stream_at: true, ..TransportParameters::default() }; params.write(&mut buf); From a03d4ffa3f90df0825ba973f17051e0eac023707 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Wed, 10 Jun 2026 20:49:50 +0200 Subject: [PATCH 2/5] Add the frame type itself --- noq-proto/src/connection/mod.rs | 6 ++ noq-proto/src/connection/qlog.rs | 12 +++ noq-proto/src/connection/stats.rs | 134 ++++++++++----------------- noq-proto/src/frame.rs | 57 +++++++++++- noq-proto/src/tests/encode_decode.rs | 1 + 5 files changed, 123 insertions(+), 87 deletions(-) diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 121a033cb3..1400ae0e56 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -5099,6 +5099,9 @@ impl Connection { self.spaces[SpaceId::Data].pending.max_data = true; } } + Frame::ResetStreamAt(frame) => { + todo!(); + } Frame::DataBlocked(DataBlocked(offset)) => { debug!(offset, "peer claims to be blocked at connection level"); } @@ -7725,6 +7728,9 @@ impl SentFrames { StreamsBlocked(streams_blocked) => { self.retransmits_mut().streams_blocked[streams_blocked.dir as usize] = true } + ResetStreamAt(reset_stream_at) => { + todo!(); + } } } } diff --git a/noq-proto/src/connection/qlog.rs b/noq-proto/src/connection/qlog.rs index 19f2f696d6..155dd60c40 100644 --- a/noq-proto/src/connection/qlog.rs +++ b/noq-proto/src/connection/qlog.rs @@ -989,6 +989,17 @@ impl ToQlog for frame::StreamMetaEncoder { } } +#[cfg(feature = "qlog")] +impl ToQlog for frame::ResetStreamAt { + fn to_qlog(&self) -> QuicFrame { + // TODO: Teach qlog about this frame type. + QuicFrame::Unknown { + frame_type_bytes: Some(self.get_type().to_u64()), + raw: None, + } + } +} + #[cfg(feature = "qlog")] impl Frame { /// Converts a [`crate::Frame`] into a [`QuicFrame`]. @@ -1077,6 +1088,7 @@ impl Frame { Self::AddAddress(f) => f.to_qlog(), Self::ReachOut(f) => f.to_qlog(), Self::RemoveAddress(f) => f.to_qlog(), + Self::ResetStreamAt(f) => f.to_qlog(), } } } diff --git a/noq-proto/src/connection/stats.rs b/noq-proto/src/connection/stats.rs index dfaf02748a..a0626bddd1 100644 --- a/noq-proto/src/connection/stats.rs +++ b/noq-proto/src/connection/stats.rs @@ -33,47 +33,95 @@ impl UdpStats { } /// Number of frames transmitted or received of each frame type. -#[derive(Default, Copy, Clone, PartialEq, Eq, derive_more::Add, derive_more::AddAssign)] +#[derive( + Default, + Copy, + Clone, + PartialEq, + Eq, + derive_more::Debug, + derive_more::Add, + derive_more::AddAssign, +)] #[non_exhaustive] #[allow(missing_docs)] pub struct FrameStats { + #[debug("ACK")] pub acks: u64, + #[debug("PATH_ACK")] pub path_acks: u64, + #[debug("ACK_FREQUENCY")] pub ack_frequency: u64, + #[debug("CRYPTO")] pub crypto: u64, + #[debug("CONNECTION_CLOSE")] pub connection_close: u64, + #[debug("DATA_BLOCKED")] pub data_blocked: u64, + #[debug("DATAGRAM")] pub datagram: u64, + #[debug("HANDSHAKE_DONE")] pub handshake_done: u8, + #[debug("IMMEDIATE_ACK")] pub immediate_ack: u64, + #[debug("MAX_DATA")] pub max_data: u64, + #[debug("MAX_STREAM_DATA")] pub max_stream_data: u64, + #[debug("MAX_STREAMS_BIDI")] pub max_streams_bidi: u64, + #[debug("MAX_STREAMS_UNI")] pub max_streams_uni: u64, + #[debug("NEW_CONNECTION_ID")] pub new_connection_id: u64, + #[debug("PATH_NEW_CONNECTION_ID")] pub path_new_connection_id: u64, + #[debug("NEW_TOKEN")] pub new_token: u64, + #[debug("PATH_CHALLENGE")] pub path_challenge: u64, + #[debug("PATH_RESPONSE")] pub path_response: u64, + #[debug("PING")] pub ping: u64, + #[debug("RESET_STREAM")] pub reset_stream: u64, + #[debug("RETIRE_CONNECTION_ID")] pub retire_connection_id: u64, + #[debug("PATH_RETIRE_CONNECTION_ID")] pub path_retire_connection_id: u64, + #[debug("STREAM_DATA_BLOCKED")] pub stream_data_blocked: u64, + #[debug("STREAMS_BLOCKED_BIDI")] pub streams_blocked_bidi: u64, + #[debug("STREAMS_BLOCKED_UNI")] pub streams_blocked_uni: u64, + #[debug("STOP_SENDING")] pub stop_sending: u64, + #[debug("STREAM")] pub stream: u64, + #[debug("OBSERVED_ADDR")] pub observed_addr: u64, + #[debug("PATH_ABANDON")] pub path_abandon: u64, + #[debug("PATH_STATUS_AVAILABLE")] pub path_status_available: u64, + #[debug("PATH_STATUS_BACKUP")] pub path_status_backup: u64, + #[debug("MAX_PATH_ID")] pub max_path_id: u64, + #[debug("PATHS_BLOCKED")] pub paths_blocked: u64, + #[debug("PATH_CIDS_BLOCKED")] pub path_cids_blocked: u64, + #[debug("ADD_ADDRESS")] pub add_address: u64, + #[debug("REACH_OUT")] pub reach_out: u64, + #[debug("REMOVE_ADDRESS")] pub remove_address: u64, + #[debug("RESET_STREAM_AT")] + pub reset_stream_at: u64, } impl FrameStats { @@ -122,93 +170,11 @@ impl FrameStats { AddIpv4Address | AddIpv6Address => inc!(add_address), ReachOutAtIpv4 | ReachOutAtIpv6 => inc!(reach_out), RemoveAddress => inc!(remove_address), + ResetStreamAt => inc!(reset_stream_at), }; } } -impl std::fmt::Debug for FrameStats { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let Self { - acks, - path_acks, - ack_frequency, - crypto, - connection_close, - data_blocked, - datagram, - handshake_done, - immediate_ack, - max_data, - max_stream_data, - max_streams_bidi, - max_streams_uni, - new_connection_id, - path_new_connection_id, - new_token, - path_challenge, - path_response, - ping, - reset_stream, - retire_connection_id, - path_retire_connection_id, - stream_data_blocked, - streams_blocked_bidi, - streams_blocked_uni, - stop_sending, - stream, - observed_addr, - path_abandon, - path_status_available, - path_status_backup, - max_path_id, - paths_blocked, - path_cids_blocked, - add_address, - reach_out, - remove_address, - } = self; - f.debug_struct("FrameStats") - .field("ACK", acks) - .field("ACK_FREQUENCY", ack_frequency) - .field("CONNECTION_CLOSE", connection_close) - .field("CRYPTO", crypto) - .field("DATA_BLOCKED", data_blocked) - .field("DATAGRAM", datagram) - .field("HANDSHAKE_DONE", handshake_done) - .field("IMMEDIATE_ACK", immediate_ack) - .field("MAX_DATA", max_data) - .field("MAX_PATH_ID", max_path_id) - .field("MAX_STREAM_DATA", max_stream_data) - .field("MAX_STREAMS_BIDI", max_streams_bidi) - .field("MAX_STREAMS_UNI", max_streams_uni) - .field("NEW_CONNECTION_ID", new_connection_id) - .field("NEW_TOKEN", new_token) - .field("PATHS_BLOCKED", paths_blocked) - .field("PATH_ABANDON", path_abandon) - .field("PATH_ACK", path_acks) - .field("PATH_STATUS_AVAILABLE", path_status_available) - .field("PATH_STATUS_BACKUP", path_status_backup) - .field("PATH_CHALLENGE", path_challenge) - .field("PATH_CIDS_BLOCKED", path_cids_blocked) - .field("PATH_NEW_CONNECTION_ID", path_new_connection_id) - .field("PATH_RESPONSE", path_response) - .field("PATH_RETIRE_CONNECTION_ID", path_retire_connection_id) - .field("PING", ping) - .field("RESET_STREAM", reset_stream) - .field("RETIRE_CONNECTION_ID", retire_connection_id) - .field("STREAM_DATA_BLOCKED", stream_data_blocked) - .field("STREAMS_BLOCKED_BIDI", streams_blocked_bidi) - .field("STREAMS_BLOCKED_UNI", streams_blocked_uni) - .field("STOP_SENDING", stop_sending) - .field("STREAM", stream) - .field("OBSERVED_ADDRESS", observed_addr) - .field("ADD_ADDRESS", add_address) - .field("REACH_OUT", reach_out) - .field("REMOVE_ADDRESS", remove_address) - .finish() - } -} - /// Statistics related to a transmission path. #[derive(Debug, Default, Copy, Clone, PartialEq, Eq)] #[non_exhaustive] diff --git a/noq-proto/src/frame.rs b/noq-proto/src/frame.rs index b7fa6c0201..f8a06e7b02 100644 --- a/noq-proto/src/frame.rs +++ b/noq-proto/src/frame.rs @@ -133,8 +133,8 @@ pub enum FrameType { ReachOutAtIpv6, #[assoc(to_u64 = 0x3d7f94)] RemoveAddress, - // #[assoc(to_u64 = 0x24)] - // ResetStreamAt, + #[assoc(to_u64 = 0x24)] + ResetStreamAt, } /// Encounter a frame ID that was not valid. @@ -219,6 +219,7 @@ pub(super) enum EncodableFrame<'a> { MaxStreamData(MaxStreamData), MaxStreams(MaxStreams), StreamsBlocked(StreamsBlocked), + ResetStreamAt(ResetStreamAt), } impl<'a> EncodableFrame<'a> { @@ -253,7 +254,8 @@ impl<'a> EncodableFrame<'a> { | EncodableFrame::MaxData(_) | EncodableFrame::MaxStreamData(_) | EncodableFrame::MaxStreams(_) - | EncodableFrame::StreamsBlocked(_) => true, + | EncodableFrame::StreamsBlocked(_) + | EncodableFrame::ResetStreamAt(_) => true, } } } @@ -465,6 +467,7 @@ pub(crate) enum Frame { AddAddress(AddAddress), ReachOut(ReachOut), RemoveAddress(RemoveAddress), + ResetStreamAt(ResetStreamAt), } impl Frame { @@ -516,6 +519,7 @@ impl Frame { AddAddress(frame) => frame.get_type(), ReachOut(frame) => frame.get_type(), RemoveAddress(_) => self::RemoveAddress::TYPE, + ResetStreamAt(_) => FrameType::ResetStreamAt, } } @@ -1709,6 +1713,18 @@ impl Iter { self.take_remaining() }, }), + FrameType::ResetStreamAt => { + let frame = ResetStreamAt { + id: self.bytes.get()?, + error_code: self.bytes.get()?, + final_offset: self.bytes.get()?, + reliable_size: self.bytes.get()?, + }; + if frame.reliable_size > frame.final_offset { + return Err(IterErr::Malformed); + } + Frame::ResetStreamAt(frame) + } }) } @@ -2494,6 +2510,41 @@ impl Encodable for RemoveAddress { } } +/// RESET_STREAM_AT frame. +/// +/// +// #[allow(unreachable_pub)] // fuzzing only +#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] +#[cfg_attr(test, derive(Arbitrary))] +#[derive(Debug, Copy, Clone, derive_more::Display)] +#[display("RESET_STREAM id: {id}")] +pub(crate) struct ResetStreamAt { + pub(crate) id: StreamId, + pub(crate) error_code: VarInt, + pub(crate) final_offset: VarInt, + pub(crate) reliable_size: VarInt, +} + +impl ResetStreamAt { + pub(crate) const fn get_type(&self) -> FrameType { + FrameType::ResetStreamAt + } +} + +impl FrameStruct for ResetStreamAt { + const SIZE_BOUND: usize = 1 + 8 + 8 + 8 + 8; +} + +impl Encodable for ResetStreamAt { + fn encode(&self, out: &mut W) { + out.write(FrameType::ResetStream); // 1 byte + out.write(self.id); // <= 8 bytes + out.write(self.error_code); // <= 8 bytes + out.write(self.final_offset); // <= 8 bytes + out.write(self.reliable_size); // <= 8 bytes + } +} + /// Helper struct for display implementations. // NOTE: Due to lifetimes in fmt::Arguments it's not possible to make this a simple function that // avoids allocations. diff --git a/noq-proto/src/tests/encode_decode.rs b/noq-proto/src/tests/encode_decode.rs index 69aa3e6ff9..f822be9085 100644 --- a/noq-proto/src/tests/encode_decode.rs +++ b/noq-proto/src/tests/encode_decode.rs @@ -58,6 +58,7 @@ fn encode_frame(frame: &Frame, buf: &mut B) { Frame::AddAddress(aa) => aa.encode(buf), Frame::ReachOut(ro) => ro.encode(buf), Frame::RemoveAddress(ra) => ra.encode(buf), + Frame::ResetStreamAt(f) => f.encode(buf), } } From c39c57eb8ad5a9a88110c16ea4d6816f944b30a3 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Wed, 10 Jun 2026 21:33:42 +0200 Subject: [PATCH 3/5] impl retransmit of RESET_STREAM_AT --- noq-proto/src/connection/spaces.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/noq-proto/src/connection/spaces.rs b/noq-proto/src/connection/spaces.rs index d6b901f1c3..4803df9c11 100644 --- a/noq-proto/src/connection/spaces.rs +++ b/noq-proto/src/connection/spaces.rs @@ -558,6 +558,7 @@ pub struct Retransmits { pub(super) remove_address: BTreeSet, /// Round and local addresses to advertise in `REACH_OUT` frames pub(super) reach_out: PendingReachOutFrames, + pub(super) reset_stream_at: Option, } impl Retransmits { @@ -584,6 +585,7 @@ impl Retransmits { add_address, remove_address, reach_out, + reset_stream_at, } = &self; !max_data && !max_stream_id.iter().any(|x| *x) @@ -608,6 +610,7 @@ impl Retransmits { && add_address.is_empty() && remove_address.is_empty() && reach_out.is_empty() + && reset_stream_at.is_none() } } @@ -635,6 +638,7 @@ impl ::std::ops::BitOrAssign for Retransmits { add_address, remove_address, mut reach_out, + reset_stream_at, } = rhs; // We reduce in-stream head-of-line blocking by queueing retransmits before other data for @@ -664,6 +668,12 @@ impl ::std::ops::BitOrAssign for Retransmits { self.add_address.extend(add_address.iter().copied()); self.remove_address.extend(remove_address.iter().copied()); self.reach_out.append(&mut reach_out); + self.reset_stream_at = match (self.reset_stream_at, reset_stream_at) { + (None, None) => None, + (None, Some(v)) => Some(v), + (Some(v), None) => Some(v), + (Some(l), Some(r)) => Some(l.min(r)), + }; } } From a4a97fdebe9e21dfd358e222440417d7da674091 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Thu, 11 Jun 2026 17:16:42 +0200 Subject: [PATCH 4/5] allow sending the RESET_STREAM_AT --- noq-proto/src/connection/mod.rs | 12 ++++--- noq-proto/src/connection/send_buffer.rs | 38 ++++++++++++++++++++- noq-proto/src/connection/spaces.rs | 12 +++---- noq-proto/src/connection/streams/mod.rs | 42 ++++++++++++++++++++++++ noq-proto/src/connection/streams/send.rs | 25 ++++++++++++++ noq-proto/src/lib.rs | 4 +-- noq/src/send_stream.rs | 31 ++++++++++++++--- 7 files changed, 145 insertions(+), 19 deletions(-) diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 1400ae0e56..44ae7d52e9 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -98,8 +98,8 @@ pub use streams::StreamsState; #[cfg(not(fuzzing))] use streams::StreamsState; pub use streams::{ - Chunks, ClosedStream, FinishError, ReadError, ReadableError, RecvStream, SendStream, - ShouldTransmit, StreamEvent, Streams, WriteError, + Chunks, ClosedStream, FinishError, ReadError, ReadableError, RecvStream, ResetStreamAtError, + SendStream, ShouldTransmit, StreamEvent, Streams, WriteError, }; mod timer; @@ -7728,8 +7728,12 @@ impl SentFrames { StreamsBlocked(streams_blocked) => { self.retransmits_mut().streams_blocked[streams_blocked.dir as usize] = true } - ResetStreamAt(reset_stream_at) => { - todo!(); + ResetStreamAt(frame) => { + self.retransmits_mut().reset_stream_at.push(( + frame.id, + frame.final_offset, + frame.error_code, + )); } } } diff --git a/noq-proto/src/connection/send_buffer.rs b/noq-proto/src/connection/send_buffer.rs index a09ddd1982..c53a759dec 100644 --- a/noq-proto/src/connection/send_buffer.rs +++ b/noq-proto/src/connection/send_buffer.rs @@ -94,6 +94,37 @@ impl SendBufferData { } } + /// Discard data from the end of the buffer. + /// + /// Calling this with offset outside of [`Self::range`] is essentially a no-op since + /// nothing needs to be truncated. + fn truncate(&mut self, offset: u64) { + if !self.range().contains(&offset) { + return; + } + + // clear truncated data + let mut n = self.offset; + for segment in self.segments.iter_mut() { + if (n + segment.len() as u64) < offset { + n += segment.len() as u64; + } else if n < offset { + segment.truncate((offset - n) as usize); + n += offset - n; + } else { + segment.clear(); + } + } + + // remove empty segments + self.segments.retain(|s| !s.is_empty()); + + // shrink segments if we have a lot of unused capacity + if self.segments.len() * 4 < self.segments.capacity() { + self.segments.shrink_to_fit(); + } + } + /// Discard data from the front of the buffer /// /// Calling this with n > len() is allowed and will simply clear the buffer. @@ -232,6 +263,11 @@ impl SendBuffer { self.retransmits.remove(0..self.fully_acked_offset()); } + pub(super) fn truncate(&mut self, offset: u64) { + self.data.truncate(offset); + self.retransmits.remove(offset..self.offset()); + } + /// Compute the next range to transmit on this stream and update state to account for that /// transmission. /// @@ -324,7 +360,7 @@ impl SendBuffer { } /// Offset up to which all data has been acknowledged - fn fully_acked_offset(&self) -> u64 { + pub(super) fn fully_acked_offset(&self) -> u64 { self.data.range().start } diff --git a/noq-proto/src/connection/spaces.rs b/noq-proto/src/connection/spaces.rs index 4803df9c11..573ace341d 100644 --- a/noq-proto/src/connection/spaces.rs +++ b/noq-proto/src/connection/spaces.rs @@ -558,7 +558,8 @@ pub struct Retransmits { pub(super) remove_address: BTreeSet, /// Round and local addresses to advertise in `REACH_OUT` frames pub(super) reach_out: PendingReachOutFrames, - pub(super) reset_stream_at: Option, + /// Pending RESET_STREAM_AT frames: (StreamId, offset, error_code). + pub(super) reset_stream_at: Vec<(StreamId, VarInt, VarInt)>, } impl Retransmits { @@ -610,7 +611,7 @@ impl Retransmits { && add_address.is_empty() && remove_address.is_empty() && reach_out.is_empty() - && reset_stream_at.is_none() + && reset_stream_at.is_empty() } } @@ -668,12 +669,7 @@ impl ::std::ops::BitOrAssign for Retransmits { self.add_address.extend(add_address.iter().copied()); self.remove_address.extend(remove_address.iter().copied()); self.reach_out.append(&mut reach_out); - self.reset_stream_at = match (self.reset_stream_at, reset_stream_at) { - (None, None) => None, - (None, Some(v)) => Some(v), - (Some(v), None) => Some(v), - (Some(l), Some(r)) => Some(l.min(r)), - }; + self.reset_stream_at.extend_from_slice(&reset_stream_at); } } diff --git a/noq-proto/src/connection/streams/mod.rs b/noq-proto/src/connection/streams/mod.rs index 4a3f49d755..471cadde74 100644 --- a/noq-proto/src/connection/streams/mod.rs +++ b/noq-proto/src/connection/streams/mod.rs @@ -356,6 +356,30 @@ impl<'a> SendStream<'a> { Ok(()) } + /// Abandon transmitting data on a stream, deliver reliably up to `offset`. + pub fn reset_at( + &mut self, + offset: VarInt, + error_code: VarInt, + ) -> Result<(), ResetStreamAtError> { + let max_send_data = self.state.max_send_data(self.id); + let stream = self + .state + .send + .get_mut(&self.id) + .map(get_or_insert_send(max_send_data)) + .ok_or(ResetStreamAtError::ClosedStream)?; + + if matches!(stream.state, SendState::ResetSent) { + return Err(ResetStreamAtError::ClosedStream); + } + stream.reset_at(offset)?; + self.pending + .reset_stream_at + .push((self.id, offset, error_code)); + Ok(()) + } + /// Set the priority of a stream /// /// # Panics @@ -542,6 +566,24 @@ impl From for io::Error { } } +/// Errors for resetting a stream with partial delivery. +#[derive(Debug, Error, Clone, PartialEq, Eq)] +pub enum ResetStreamAtError { + /// The stream has already been stopped. + /// + /// The peer is no longer accepting data on this stream. + /// + /// Carries an application-defined error code. + #[error("stopped by peer: code {0}")] + Stopped(VarInt), + /// The stream has already been finished or reset. + #[error("closed stream")] + ClosedStream, + /// The reliable size is larger than the number of bytes sent on the stream. + #[error("invalid reliable size")] + InvalidReliableSize, +} + #[derive(Debug, Copy, Clone, Eq, PartialEq)] enum StreamHalf { Send, diff --git a/noq-proto/src/connection/streams/send.rs b/noq-proto/src/connection/streams/send.rs index cc30c2c8d2..f99202a595 100644 --- a/noq-proto/src/connection/streams/send.rs +++ b/noq-proto/src/connection/streams/send.rs @@ -7,6 +7,8 @@ use crate::{ frame, }; +use super::ResetStreamAtError; + #[derive(Debug)] pub(super) struct Send { pub(super) max_data: u64, @@ -53,6 +55,29 @@ impl Send { } } + pub(super) fn reset_at(&mut self, offset: VarInt) -> Result<(), ResetStreamAtError> { + if let Some(error_code) = self.stop_reason { + return Err(ResetStreamAtError::Stopped(error_code)); + } + match self.state { + SendState::Ready | SendState::DataSent { .. } if offset.0 >= self.pending.offset() => { + Err(ResetStreamAtError::InvalidReliableSize) + } + SendState::Ready => { + self.state = SendState::DataSent { + finish_acked: false, + }; + self.pending.truncate(offset.0); + Ok(()) + } + SendState::DataSent { .. } => { + self.pending.truncate(offset.0); + Ok(()) + } + SendState::ResetSent => Err(ResetStreamAtError::ClosedStream), + } + } + pub(super) fn write<'a, S: BytesSource<'a>>( &mut self, source: &'a mut S, diff --git a/noq-proto/src/lib.rs b/noq-proto/src/lib.rs index 85bfdcf9e8..7d6a181bce 100644 --- a/noq-proto/src/lib.rs +++ b/noq-proto/src/lib.rs @@ -45,8 +45,8 @@ pub use crate::connection::{ Chunk, Chunks, ClosePathError, ClosedPath, ClosedStream, Connection, ConnectionError, ConnectionStats, Datagrams, Event, FinishError, FrameStats, MultipathNotNegotiated, NetworkChangeHint, PathAbandonReason, PathError, PathEvent, PathId, PathStats, PathStatus, - ReadError, ReadableError, RecvStream, RttEstimator, SendDatagramError, SendStream, - SetPathStatusError, ShouldTransmit, StreamEvent, Streams, UdpStats, WriteError, + ReadError, ReadableError, RecvStream, ResetStreamAtError, RttEstimator, SendDatagramError, + SendStream, SetPathStatusError, ShouldTransmit, StreamEvent, Streams, UdpStats, WriteError, }; #[cfg(test)] use test_strategy::Arbitrary; diff --git a/noq/src/send_stream.rs b/noq/src/send_stream.rs index a46771a012..534e32d853 100644 --- a/noq/src/send_stream.rs +++ b/noq/src/send_stream.rs @@ -7,7 +7,7 @@ use std::{ use bytes::Bytes; use pin_project_lite::pin_project; -use proto::{ClosedStream, ConnectionError, FinishError, StreamId}; +use proto::{ClosedStream, ConnectionError, FinishError, ResetStreamAtError, StreamId}; use thiserror::Error; use tokio::sync::futures::OwnedNotified; @@ -79,7 +79,7 @@ impl SendStream { Ok(()) } - /// Writes [`Bytes`] from a slice of buffers into this stream, returning how many bytes were. + /// Writes [`Bytes`] from a slice of buffers into this stream, returning how many bytes were. /// written /// /// Bytes to try to write are provided to this method as an array of cheaply cloneable chunks. @@ -103,7 +103,7 @@ impl SendStream { poll_fn(|cx| self.execute_poll(cx, |s| s.write_chunks(bufs))).await } - /// Writes a single [`Bytes`] into this stream in its entirety. + /// Writes a single [`Bytes`] into this stream in its entirety. /// /// Bytes to write are provided to this method as a single cheaply cloneable chunk. This /// method repeatedly calls [`write_many_chunks`](Self::write_many_chunks) until all bytes @@ -117,7 +117,7 @@ impl SendStream { self.write_all_chunks(&mut [buf]).await } - /// Writes a slice of [`Bytes`] into this stream in its entirety. + /// Writes a slice of [`Bytes`] into this stream in its entirety. /// /// Bytes to write are provided to this method as an array of cheaply cloneable chunks. This /// method repeatedly calls [`write_many_chunks`](Self::write_many_chunks) until all bytes are @@ -219,6 +219,29 @@ impl SendStream { Ok(()) } + /// Close the send stream immediately, delivering data up to `offset`. + /// + /// No new data can be written after calling this method. Data up to `offset` is still + /// reliably delivered, but any remaining stream data may never be transmitted or lost. + /// + /// Fails if [`Self::finish`], [`Self::reset`] or [`Self::reset_at`] was previously + /// called, of if the remote stopped the stream. + pub fn reset_at( + &mut self, + offset: VarInt, + error_code: VarInt, + ) -> Result<(), ResetStreamAtError> { + let mut conn = self.conn.lock_and_wake("SendStream::reset_at"); + if self.is_0rtt && conn.check_0rtt().is_err() { + conn.skip_waking(); + return Ok(()); + } + conn.inner + .send_stream(self.stream) + .reset_at(offset, error_code)?; + Ok(()) + } + /// Set the priority of the send stream /// /// Every send stream has an initial priority of 0. Locally buffered data from streams with From ca4b44552c6df355906f45c58e992ddbb7986bbb Mon Sep 17 00:00:00 2001 From: dignifiedquire Date: Mon, 15 Jun 2026 11:11:42 +0200 Subject: [PATCH 5/5] wip: thanks to claude --- noq-proto/src/connection/assembler.rs | 21 + noq-proto/src/connection/mod.rs | 25 +- noq-proto/src/connection/send_buffer.rs | 154 +++++- noq-proto/src/connection/spaces.rs | 11 +- noq-proto/src/connection/streams/mod.rs | 59 ++- noq-proto/src/connection/streams/recv.rs | 328 +++++++++++- noq-proto/src/connection/streams/send.rs | 233 ++++++++- noq-proto/src/connection/streams/state.rs | 599 +++++++++++++++++++++- noq-proto/src/frame.rs | 18 +- noq-proto/src/tests/encode_decode.rs | 51 ++ noq-proto/src/transport_parameters.rs | 27 +- noq/src/lib.rs | 8 +- noq/src/send_stream.rs | 24 +- noq/src/tests.rs | 54 ++ 14 files changed, 1537 insertions(+), 75 deletions(-) diff --git a/noq-proto/src/connection/assembler.rs b/noq-proto/src/connection/assembler.rs index 16a40638d6..d910ea0cb0 100644 --- a/noq-proto/src/connection/assembler.rs +++ b/noq-proto/src/connection/assembler.rs @@ -58,6 +58,20 @@ impl Assembler { /// Get the the next chunk pub(super) fn read(&mut self, max_length: usize, ordered: bool) -> Option { + self.read_capped(max_length, ordered, u64::MAX) + } + + /// Like [`Self::read`], but never returns data at or beyond `offset_limit`. + /// + /// Used to deliver only the reliable prefix of a stream subject to a RESET_STREAM_AT. The limit + /// is applied by stream offset rather than by the read cursor, because in unordered mode the + /// cursor is a running total of bytes handed out, not the contiguous prefix offset. + pub(super) fn read_capped( + &mut self, + max_length: usize, + ordered: bool, + offset_limit: u64, + ) -> Option { loop { let mut chunk = self.data.peek_mut()?; @@ -82,6 +96,13 @@ impl Assembler { } } + // Never hand out data at or beyond the offset limit, and never let a single chunk + // straddle it. + if chunk.offset >= offset_limit { + return None; + } + let max_length = max_length.min((offset_limit - chunk.offset) as usize); + return Some(if max_length < chunk.bytes.len() { self.bytes_read += max_length as u64; let offset = chunk.offset; diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index 44ae7d52e9..5278b0e80c 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -3152,6 +3152,9 @@ impl Connection { for (id, _) in retransmits.reset_stream.iter() { self.streams.reset_acked(*id); } + for (id, reliable_size) in retransmits.reset_stream_at.iter() { + self.streams.reset_at_acked(*id, *reliable_size); + } } for frame in info.stream_frames { @@ -5100,7 +5103,16 @@ impl Connection { } } Frame::ResetStreamAt(frame) => { - todo!(); + // We only advertise (and thus permit) RESET_STREAM_AT when the extension is + // enabled; receiving one otherwise is a protocol violation. + if !self.endpoint_config.reset_stream_at { + return Err(TransportError::PROTOCOL_VIOLATION( + "RESET_STREAM_AT frame received without negotiating the extension", + )); + } + if self.streams.received_reset_at(frame)?.should_transmit() { + self.spaces[SpaceId::Data].pending.max_data = true; + } } Frame::DataBlocked(DataBlocked(offset)) => { debug!(offset, "peer claims to be blocked at connection level"); @@ -7728,13 +7740,10 @@ impl SentFrames { StreamsBlocked(streams_blocked) => { self.retransmits_mut().streams_blocked[streams_blocked.dir as usize] = true } - ResetStreamAt(frame) => { - self.retransmits_mut().reset_stream_at.push(( - frame.id, - frame.final_offset, - frame.error_code, - )); - } + ResetStreamAt(frame) => self + .retransmits_mut() + .reset_stream_at + .push((frame.id, frame.reliable_size)), } } } diff --git a/noq-proto/src/connection/send_buffer.rs b/noq-proto/src/connection/send_buffer.rs index c53a759dec..6c24274fe5 100644 --- a/noq-proto/src/connection/send_buffer.rs +++ b/noq-proto/src/connection/send_buffer.rs @@ -94,27 +94,37 @@ impl SendBufferData { } } - /// Discard data from the end of the buffer. + /// Discard data from the end of the buffer, retaining only the bytes before `offset`. /// - /// Calling this with offset outside of [`Self::range`] is essentially a no-op since - /// nothing needs to be truncated. + /// `offset` is an absolute stream offset. Values at or beyond [`Self::range`]`.end` are a + /// no-op (nothing to drop); values at or below [`Self::range`]`.start` clear the buffer. The + /// front offset ([`Self::range`]`.start`) is never moved, so already-acknowledged-and-popped + /// data is unaffected. fn truncate(&mut self, offset: u64) { - if !self.range().contains(&offset) { + // Number of bytes to retain from the front of the buffer. + let new_len = offset.saturating_sub(self.offset).min(self.len as u64) as usize; + if new_len >= self.len { return; } - // clear truncated data - let mut n = self.offset; + // Walk the stored segments (and finally `last_segment`), keeping the first `new_len` + // bytes and dropping everything after them. + let mut kept = 0usize; for segment in self.segments.iter_mut() { - if (n + segment.len() as u64) < offset { - n += segment.len() as u64; - } else if n < offset { - segment.truncate((offset - n) as usize); - n += offset - n; - } else { + if kept >= new_len { segment.clear(); + } else if kept + segment.len() > new_len { + segment.truncate(new_len - kept); + kept = new_len; + } else { + kept += segment.len(); } } + // Any remainder lives in `last_segment`. If the segments already cover `new_len`, the + // whole `last_segment` is beyond the cut and is dropped. + self.last_segment.truncate(new_len.saturating_sub(kept)); + + self.len = new_len; // remove empty segments self.segments.retain(|s| !s.is_empty()); @@ -263,9 +273,22 @@ impl SendBuffer { self.retransmits.remove(0..self.fully_acked_offset()); } + /// Discard buffered data at or beyond `offset`, abandoning it. + /// + /// Used to implement RESET_STREAM_AT: data past the reliable size is no longer (re)transmitted. + /// The new end offset is clamped to the fully-acknowledged offset, so already-sent-and-acked + /// data is retained. Bookkeeping for the discarded tail (unsent cursor, queued retransmits, and + /// acknowledged ranges) is dropped to keep the buffer consistent. pub(super) fn truncate(&mut self, offset: u64) { + let old_end = self.offset(); self.data.truncate(offset); - self.retransmits.remove(offset..self.offset()); + let new_end = self.offset(); + debug_assert!(new_end <= old_end); + + // We no longer have the discarded data, so we must not try to send or track it. + self.unsent = self.unsent.min(new_end); + self.retransmits.remove(new_end..old_end); + self.acks.remove(new_end..old_end); } /// Compute the next range to transmit on this stream and update state to account for that @@ -597,6 +620,94 @@ mod tests { assert!(buf.acks.is_empty()); } + #[test] + fn truncate_basic() { + let mut buf = SendBuffer::new(); + const MSG: &[u8] = b"Hello, world!"; // 13 bytes, coalesced into last_segment + buf.write(MSG); + assert_eq!(buf.offset(), 13); + + buf.truncate(5); + assert_eq!(buf.offset(), 5); + assert_eq!(aggregate_unacked(&buf), &MSG[..5]); + + // Truncating at or past the end is a no-op. + buf.truncate(100); + assert_eq!(buf.offset(), 5); + assert_eq!(aggregate_unacked(&buf), &MSG[..5]); + + // poll_transmit never yields data beyond the truncation point. + assert_eq!(buf.poll_transmit(64), (0..5, true)); + assert_eq!(buf.poll_transmit(64), (5..5, true)); + assert!(!buf.has_unsent_data()); + } + + #[test] + fn truncate_multiple_segments() { + let mut buf = SendBuffer::new(); + // Segments larger than MAX_COMBINE are stored as standalone segments. + buf.write(Bytes::from(vec![1u8; 2000])); + buf.write(Bytes::from(vec![2u8; 2000])); + assert_eq!(buf.offset(), 4000); + + // Cut inside the second segment. + buf.truncate(3000); + assert_eq!(buf.offset(), 3000); + let data = aggregate_unacked(&buf); + assert_eq!(data.len(), 3000); + assert!(data[..2000].iter().all(|&x| x == 1)); + assert!(data[2000..].iter().all(|&x| x == 2)); + + // Cut exactly at the boundary of the first segment. + buf.truncate(2000); + assert_eq!(buf.offset(), 2000); + let data = aggregate_unacked(&buf); + assert_eq!(data.len(), 2000); + assert!(data.iter().all(|&x| x == 1)); + } + + #[test] + fn truncate_after_ack_and_below_front() { + let mut buf = SendBuffer::new(); + const MSG: &[u8] = b"abcdefghij"; // 10 bytes + buf.write(MSG); + assert_eq!(buf.poll_transmit(64), (0..10, true)); + buf.ack(0..4); // drop the first 4 bytes from the front + assert_eq!(buf.fully_acked_offset(), 4); + + // Truncate within the retained range. + buf.truncate(7); + assert_eq!(buf.offset(), 7); + assert_eq!(aggregate_unacked(&buf), b"efg"); + + // Truncating below the (advanced) front clears the buffer but keeps the front offset. + buf.truncate(2); + assert_eq!(buf.offset(), 4); + assert_eq!(buf.fully_acked_offset(), 4); + assert!(buf.is_fully_acked()); + assert!(aggregate_unacked(&buf).is_empty()); + } + + #[test] + fn truncate_drops_retransmits_and_unacked() { + let mut buf = SendBuffer::new(); + const MSG: &[u8] = b"Hello, world with extra data!"; // 29 bytes + buf.write(MSG); + assert_eq!(buf.poll_transmit(64), (0..29, true)); + // Mark a tail range lost, queuing it for retransmission. + buf.retransmit(20..29); + assert_eq!(buf.unacked(), 29); + assert!(buf.has_unsent_data()); + + // Truncating away the lost tail must drop the queued retransmit and update accounting, + // otherwise a later poll_transmit would request data we no longer have and panic. + buf.truncate(15); + assert_eq!(buf.offset(), 15); + assert_eq!(buf.unacked(), 15); + assert!(!buf.has_unsent_data()); + assert_eq!(buf.poll_transmit(64), (15..15, true)); + } + fn aggregate_unacked(buf: &SendBuffer) -> Vec { buf.data.to_vec() } @@ -636,6 +747,8 @@ mod proptests { Retransmit(Range), // poll_transmit with the given max len PollTransmit(#[strategy(16usize..1024)] usize), + // truncate the buffer at a random offset within the sent range + Truncate(u64), } /// Map a range into a target range @@ -690,6 +803,21 @@ mod proptests { trace!("Op::Retransmit({:?})", range); sb.retransmit(range); } + Op::Truncate(raw) => { + // Choose a truncation point within the data that has actually been sent. + let target = if max_send_offset == 0 { + 0 + } else { + raw % (max_send_offset + 1) + }; + trace!("Op::Truncate({})", target); + sb.truncate(target); + // `truncate` clamps the new end to the fully-acked offset, so read back the + // authoritative end and mirror it in the reference model. + let new_end = sb.offset(); + buf.truncate(new_end as usize); + max_send_offset = max_send_offset.min(new_end); + } Op::PollTransmit(max_len) => { trace!("Op::PollTransmit({})", max_len); let (range, _partial) = sb.poll_transmit(max_len); diff --git a/noq-proto/src/connection/spaces.rs b/noq-proto/src/connection/spaces.rs index 573ace341d..ec703254a8 100644 --- a/noq-proto/src/connection/spaces.rs +++ b/noq-proto/src/connection/spaces.rs @@ -558,8 +558,15 @@ pub struct Retransmits { pub(super) remove_address: BTreeSet, /// Round and local addresses to advertise in `REACH_OUT` frames pub(super) reach_out: PendingReachOutFrames, - /// Pending RESET_STREAM_AT frames: (StreamId, offset, error_code). - pub(super) reset_stream_at: Vec<(StreamId, VarInt, VarInt)>, + /// Streams that need a RESET_STREAM_AT frame (re)transmitted, paired with the reliable size the + /// pending/sent frame carries. + /// + /// The final size, reliable size, and error code of an outgoing frame are reconstructed from + /// the live send-stream state when it is written (mirroring how `reset_stream` rebuilds its + /// final offset), so the stored reliable size is ignored on (re)transmission. It is retained + /// only so that, on acknowledgement, a frame carrying a now-superseded (larger) reliable size + /// can be distinguished from the current one (see `reset_at_acked`). + pub(super) reset_stream_at: Vec<(StreamId, VarInt)>, } impl Retransmits { diff --git a/noq-proto/src/connection/streams/mod.rs b/noq-proto/src/connection/streams/mod.rs index 471cadde74..ec01e7709f 100644 --- a/noq-proto/src/connection/streams/mod.rs +++ b/noq-proto/src/connection/streams/mod.rs @@ -15,7 +15,7 @@ use crate::{ }; mod recv; -use recv::Recv; +use recv::{Recv, ResetAtOutcome}; pub use recv::{Chunks, ReadError, ReadableError}; mod send; @@ -340,8 +340,8 @@ impl<'a> SendStream<'a> { .map(get_or_insert_send(max_send_data)) .ok_or(ClosedStream { _private: () })?; - if matches!(stream.state, SendState::ResetSent) { - // Redundant reset call + if matches!(stream.state, SendState::ResetSent) || stream.reset_at.is_some() { + // Redundant reset call, or a reliable reset (RESET_STREAM_AT) is already in progress. return Err(ClosedStream { _private: () }); } @@ -356,12 +356,32 @@ impl<'a> SendStream<'a> { Ok(()) } - /// Abandon transmitting data on a stream, deliver reliably up to `offset`. + /// Abandon transmitting data on a stream, reliably delivering data up to `reliable_size` first. + /// + /// Sends a RESET_STREAM_AT frame ([draft-ietf-quic-reliable-stream-reset]): the peer is + /// delivered all stream data up to `reliable_size` and only then observes the reset carrying + /// `error_code`. Unlike [`Self::reset`], data up to the reliable size is retransmitted on loss. + /// + /// May be called repeatedly to *reduce* the reliable size (the error code must stay the same). + /// `reliable_size` may not exceed the number of bytes written to the stream. + /// + /// Returns [`ResetStreamAtError::Unsupported`] if the peer did not advertise support for the + /// extension, in which case no frame is sent and the stream is left untouched. + /// + /// [draft-ietf-quic-reliable-stream-reset]: https://datatracker.ietf.org/doc/html/draft-ietf-quic-reliable-stream-reset + /// + /// # Panics + /// - when applied to a receive stream pub fn reset_at( &mut self, - offset: VarInt, + reliable_size: VarInt, error_code: VarInt, ) -> Result<(), ResetStreamAtError> { + // We may only send RESET_STREAM_AT frames if the peer advertised that it can receive them. + if !self.state.peer_reset_stream_at { + return Err(ResetStreamAtError::Unsupported); + } + let max_send_data = self.state.max_send_data(self.id); let stream = self .state @@ -370,13 +390,22 @@ impl<'a> SendStream<'a> { .map(get_or_insert_send(max_send_data)) .ok_or(ResetStreamAtError::ClosedStream)?; - if matches!(stream.state, SendState::ResetSent) { - return Err(ResetStreamAtError::ClosedStream); + // Restore the send window consumed by data beyond the reliable size, which we are about to + // discard. Connection-level flow control is left to the peer, which reissues credit based + // on the final size communicated in the RESET_STREAM_AT frame. + let unacked_before = stream.pending.unacked(); + let queue_frame = stream.reset_at(reliable_size, error_code)?; + // The committed reliable size is the (possibly clamped) truncated send-buffer end. + let committed_reliable = + VarInt::try_from(stream.pending.offset()).expect("offset fits in varint"); + let unacked_after = stream.pending.unacked(); + self.state.unacked_data -= unacked_before - unacked_after; + + if queue_frame { + self.pending + .reset_stream_at + .push((self.id, committed_reliable)); } - stream.reset_at(offset)?; - self.pending - .reset_stream_at - .push((self.id, offset, error_code)); Ok(()) } @@ -579,9 +608,15 @@ pub enum ResetStreamAtError { /// The stream has already been finished or reset. #[error("closed stream")] ClosedStream, - /// The reliable size is larger than the number of bytes sent on the stream. + /// The requested reliable reset is invalid given the stream's state: the reliable size exceeds + /// the bytes written, exceeds a previously committed reliable size, or the error code differs + /// from an earlier RESET_STREAM_AT for the same stream. #[error("invalid reliable size")] InvalidReliableSize, + /// The peer did not advertise support for receiving RESET_STREAM_AT frames, so a reliable reset + /// cannot be performed. + #[error("peer does not support reliable reset")] + Unsupported, } #[derive(Debug, Copy, Clone, Eq, PartialEq)] diff --git a/noq-proto/src/connection/streams/recv.rs b/noq-proto/src/connection/streams/recv.rs index afa5b3e1dd..62b10bd03a 100644 --- a/noq-proto/src/connection/streams/recv.rs +++ b/noq-proto/src/connection/streams/recv.rs @@ -153,14 +153,20 @@ impl Recv { } /// Whether data is still being accepted from the peer + /// + /// Remains true after a RESET_STREAM_AT frame is received, since the sender may still + /// retransmit reliable data needed to fill gaps below the reliable size. pub(super) fn is_receiving(&self) -> bool { - matches!(self.state, RecvState::Recv { .. }) + matches!( + self.state, + RecvState::Recv { .. } | RecvState::ResetRecvdAt { .. } + ) } fn final_offset(&self) -> Option { match self.state { RecvState::Recv { size } => size, - RecvState::ResetRecvd { size, .. } => Some(size), + RecvState::ResetRecvd { size, .. } | RecvState::ResetRecvdAt { size, .. } => Some(size), } } @@ -199,6 +205,108 @@ impl Recv { Ok(true) } + /// Process a RESET_STREAM_AT (reliable reset) frame. + /// + /// `final_size` is the stream's final size and `reliable_size` (`<= final_size`) the offset up + /// to which data must still be delivered to the application. The error code is surfaced to the + /// application only after it has read all data up to the reliable size. + /// + /// Establishes the final size for flow control like [`Self::reset`], but instead of discarding + /// buffered data it retains it for delivery up to the reliable size. To avoid double-counting + /// connection flow control as retransmitted reliable data arrives, the receive high-water mark + /// is advanced to the final size here, so subsequent [`Self::ingest`] calls consume no further + /// credit. + pub(super) fn reset_at( + &mut self, + error_code: VarInt, + final_size: VarInt, + reliable_size: VarInt, + received: u64, + max_data: u64, + ) -> Result { + let final_size = final_size.into_inner(); + // The wire decoder already rejects `reliable > final`; clamp defensively for callers that + // construct frames directly (e.g. fuzzing). + let reliable_size = reliable_size.into_inner().min(final_size); + + // The final size is immutable once known and may not be below already-received data. + if let Some(known) = self.final_offset() { + if known != final_size { + return Err(TransportError::FINAL_SIZE_ERROR("inconsistent value")); + } + } else if self.end > final_size { + return Err(TransportError::FINAL_SIZE_ERROR( + "lower than high water mark", + )); + } + + // We cannot un-deliver data already read beyond the reliable size. + let deliver_cap = reliable_size.max(self.assembler.bytes_read()); + + match self.state { + RecvState::ResetRecvd { + error_code: prev_code, + .. + } => { + // The error code is immutable across frames for the same stream (RFC §5.2). + if error_code != prev_code { + return Err(TransportError::STREAM_STATE_ERROR( + "RESET_STREAM_AT error code changed", + )); + } + // An ordinary reset already fully terminated the stream (reliable size 0); a + // reliable reset cannot un-terminate it or reduce the reliable size below zero. + Ok(ResetAtOutcome::Ignored) + } + RecvState::ResetRecvdAt { + reliable_size: prev_cap, + error_code: prev_code, + .. + } => { + // The error code is immutable across frames for the same stream (RFC §5.2). + if error_code != prev_code { + return Err(TransportError::STREAM_STATE_ERROR( + "RESET_STREAM_AT error code changed", + )); + } + if deliver_cap >= prev_cap { + // The reliable size did not decrease; the spec requires ignoring increases. + return Ok(ResetAtOutcome::Ignored); + } + self.state = RecvState::ResetRecvdAt { + size: final_size, + reliable_size: deliver_cap, + error_code, + }; + // Release connection credit for data between the old and new caps that will no + // longer be delivered. The final size, and thus `data_recvd`, is unchanged. + Ok(ResetAtOutcome::Applied { + received_delta: 0, + credit: prev_cap - deliver_cap, + }) + } + RecvState::Recv { .. } => { + // Bounds-check the final size against flow control. The returned value is the + // number of so-far-unaccounted bytes up to the final size (`final_size - end`). + let received_delta = self.credit_consumed_by(final_size, received, max_data)?; + // Account every byte up to the final size as received, so retransmitted reliable + // data ingested later adds no further credit (see method docs). + self.end = final_size; + self.state = RecvState::ResetRecvdAt { + size: final_size, + reliable_size: deliver_cap, + error_code, + }; + // Release connection credit for the tail beyond the reliable size, which will never + // be delivered. Data up to the reliable size releases its credit as it is read. + Ok(ResetAtOutcome::Applied { + received_delta, + credit: final_size - deliver_cap, + }) + } + } + } + pub(super) fn reset_code(&self) -> Option { match self.state { RecvState::ResetRecvd { error_code, .. } => Some(error_code), @@ -206,6 +314,19 @@ impl Recv { } } + /// If a reliable reset (RESET_STREAM_AT) is in progress, the offset up to which data is still to + /// be delivered to the application. + /// + /// Used when a plain RESET_STREAM follows a RESET_STREAM_AT: only the credit for the + /// still-deliverable region `[bytes_read, cap)` must be released, because the tail beyond the + /// reliable size already had its credit released when the RESET_STREAM_AT was processed. + pub(super) fn reliable_reset_deliver_cap(&self) -> Option { + match self.state { + RecvState::ResetRecvdAt { reliable_size, .. } => Some(reliable_size), + _ => None, + } + } + /// Compute the amount of flow control credit consumed, or return an error if more was consumed /// than issued fn credit_consumed_by( @@ -296,7 +417,16 @@ impl<'a> Chunks<'a> { ChunksState::Finalized => panic!("must not call next() after finalize()"), }; - if let Some(chunk) = rs.assembler.read(max_length, self.ordered) { + // For a reliable reset, never deliver data beyond the reliable size. The cap is by stream + // offset (not the read cursor) so it is correct for unordered reads too. + let chunk = match rs.state { + RecvState::ResetRecvdAt { reliable_size, .. } => { + rs.assembler + .read_capped(max_length, self.ordered, reliable_size) + } + _ => rs.assembler.read(max_length, self.ordered), + }; + if let Some(chunk) = chunk { self.read += chunk.bytes.len() as u64; return Ok(Some(chunk)); } @@ -313,6 +443,26 @@ impl<'a> Chunks<'a> { self.streams.stream_recv_freed(self.id, recv); Err(ReadError::Reset(error_code)) } + RecvState::ResetRecvdAt { + reliable_size, + error_code, + .. + } => { + if rs.assembler.bytes_read() >= reliable_size { + // All reliable data has been delivered; surface the reset and dispose of the + // stream. Any data buffered beyond the reliable size is dropped undelivered. + let state = mem::replace(&mut self.state, ChunksState::Reset(error_code)); + let recv = match state { + ChunksState::Readable(recv) => StreamRecv::Open(recv), + _ => unreachable!("state must be ChunkState::Readable"), + }; + self.streams.stream_recv_freed(self.id, recv); + Err(ReadError::Reset(error_code)) + } else { + // A gap remains below the reliable size; wait for the sender to retransmit it. + Err(ReadError::Blocked) + } + } RecvState::Recv { size } => { if size == Some(rs.end) && rs.assembler.bytes_read() == rs.end { let state = mem::replace(&mut self.state, ChunksState::Finished); @@ -429,10 +579,41 @@ impl From for ReadableError { } } +/// The effect of a RESET_STREAM_AT frame on connection-level flow control, returned by +/// [`Recv::reset_at`] for the caller to apply. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum ResetAtOutcome { + /// The frame was redundant, or attempted to increase the reliable size, and was ignored. + Ignored, + /// A reliable reset was newly established or its reliable size reduced. + Applied { + /// Amount to add to the connection-level received-bytes counter (`data_recvd`): the + /// never-arriving tail up to the final size. Zero when only reducing the reliable size. + received_delta: u64, + /// Connection flow-control credit to release immediately for data beyond the reliable size + /// that will never be delivered to the application. + credit: u64, + }, +} + #[derive(Debug, Copy, Clone, Eq, PartialEq)] enum RecvState { - Recv { size: Option }, - ResetRecvd { size: u64, error_code: VarInt }, + Recv { + size: Option, + }, + ResetRecvd { + size: u64, + error_code: VarInt, + }, + /// A RESET_STREAM_AT (reliable reset) was received. Stream data up to `reliable_size` is still + /// delivered to the application; once it has all been read, the application observes the reset + /// carrying `error_code`. `size` is the stream's final size (`>= reliable_size`), used for + /// flow-control accounting and final-size consistency checks. + ResetRecvdAt { + size: u64, + reliable_size: u64, + error_code: VarInt, + }, } impl Default for RecvState { @@ -541,4 +722,141 @@ mod tests { "stream flow control credit isn't issued after stop" ); } + + const WINDOW: u64 = 1024; + + fn stream_id() -> StreamId { + StreamId::new(Side::Client, Dir::Uni, 0) + } + + /// Ingest `len` bytes at `offset` into a receive stream, returning the new connection bytes. + fn ingest_at(r: &mut Recv, offset: u64, len: usize, received: u64) -> u64 { + let data = Bytes::from(vec![0u8; len]); + let (new_bytes, _) = r + .ingest( + frame::Stream { + id: stream_id(), + offset, + fin: false, + data, + }, + len, + received, + WINDOW, + ) + .unwrap(); + new_bytes + } + + #[test] + fn reset_at_establishes_reliable_reset() { + let mut r = Recv::new(WINDOW); + assert_eq!(ingest_at(&mut r, 0, 30, 0), 30); + + // Final size 100, deliver up to 40. + let outcome = r + .reset_at(7u32.into(), 100u32.into(), 40u32.into(), 30, WINDOW) + .unwrap(); + // The never-arriving tail [30, 100) is accounted as received (70 bytes); the never-delivered + // tail [40, 100) releases its connection credit immediately (60 bytes). + assert_eq!( + outcome, + ResetAtOutcome::Applied { + received_delta: 70, + credit: 60, + } + ); + assert!( + r.is_receiving(), + "must keep accepting retransmitted reliable data" + ); + assert_eq!(r.reset_code(), None, "reset surfaces only after delivery"); + } + + #[test] + fn reset_at_does_not_double_count_retransmits() { + let mut r = Recv::new(WINDOW); + assert_eq!(ingest_at(&mut r, 0, 20, 0), 20); + r.reset_at(0u32.into(), 100u32.into(), 40u32.into(), 20, WINDOW) + .unwrap(); + + // Retransmitted reliable data [20, 40) arriving after the reset must not consume any more + // connection credit: the final size was already fully accounted. + assert_eq!(ingest_at(&mut r, 20, 20, 100), 0); + } + + #[test] + fn reset_at_ignores_reliable_size_increase() { + let mut r = Recv::new(WINDOW); + r.reset_at(0u32.into(), 100u32.into(), 40u32.into(), 0, WINDOW) + .unwrap(); + // A later (reordered) frame raising the reliable size must be ignored. + let outcome = r + .reset_at(0u32.into(), 100u32.into(), 60u32.into(), 100, WINDOW) + .unwrap(); + assert_eq!(outcome, ResetAtOutcome::Ignored); + } + + #[test] + fn reset_at_reduces_reliable_size() { + let mut r = Recv::new(WINDOW); + r.reset_at(0u32.into(), 100u32.into(), 40u32.into(), 0, WINDOW) + .unwrap(); + // Reducing 40 -> 25 releases the 15 bytes that will no longer be delivered. + let outcome = r + .reset_at(0u32.into(), 100u32.into(), 25u32.into(), 100, WINDOW) + .unwrap(); + assert_eq!( + outcome, + ResetAtOutcome::Applied { + received_delta: 0, + credit: 15, + } + ); + } + + #[test] + fn reset_at_final_size_must_be_consistent() { + let mut r = Recv::new(WINDOW); + r.reset_at(0u32.into(), 100u32.into(), 40u32.into(), 0, WINDOW) + .unwrap(); + let err = r + .reset_at(0u32.into(), 90u32.into(), 40u32.into(), 100, WINDOW) + .unwrap_err(); + assert_eq!(err.code, crate::TransportErrorCode::FINAL_SIZE_ERROR); + } + + #[test] + fn reset_at_final_size_below_high_water_is_error() { + let mut r = Recv::new(WINDOW); + assert_eq!(ingest_at(&mut r, 0, 50, 0), 50); + // A final size below the data already received is illegal. + let err = r + .reset_at(0u32.into(), 40u32.into(), 30u32.into(), 50, WINDOW) + .unwrap_err(); + assert_eq!(err.code, crate::TransportErrorCode::FINAL_SIZE_ERROR); + } + + #[test] + fn reset_at_respects_connection_flow_control() { + let mut r = Recv::new(WINDOW); + // Final size 50 would push consumption (20 already + 50) past the 40-byte budget. + let err = r + .reset_at(0u32.into(), 50u32.into(), 30u32.into(), 20, 40) + .unwrap_err(); + assert_eq!(err.code, crate::TransportErrorCode::FLOW_CONTROL_ERROR); + } + + #[test] + fn reset_at_error_code_is_immutable() { + let mut r = Recv::new(WINDOW); + r.reset_at(7u32.into(), 100u32.into(), 40u32.into(), 0, WINDOW) + .unwrap(); + // A later frame for the same stream that changes the error code is a STREAM_STATE_ERROR + // (RFC §5.2), even when it does not reduce the reliable size. + let err = r + .reset_at(8u32.into(), 100u32.into(), 40u32.into(), 100, WINDOW) + .unwrap_err(); + assert_eq!(err.code, crate::TransportErrorCode::STREAM_STATE_ERROR); + } } diff --git a/noq-proto/src/connection/streams/send.rs b/noq-proto/src/connection/streams/send.rs index f99202a595..fc54650503 100644 --- a/noq-proto/src/connection/streams/send.rs +++ b/noq-proto/src/connection/streams/send.rs @@ -21,6 +21,25 @@ pub(super) struct Send { pub(super) connection_blocked: bool, /// The reason the peer wants us to stop, if `STOP_SENDING` was received pub(super) stop_reason: Option, + /// State of an in-progress reliable reset (RESET_STREAM_AT), if any. + /// + /// When set, the stream is in [`SendState::DataSent`] and keeps (re)transmitting buffered data + /// up to the reliable size (the send buffer is truncated to it) before being closed with the + /// reset's error code. Distinguishes a reliable reset from an ordinary FIN-based finish. + pub(super) reset_at: Option, +} + +/// State for an in-progress reliable reset, see [`Send::reset_at`]. +#[derive(Debug)] +pub(super) struct ResetAt { + /// The stream's final size: the send offset captured at the first `reset_at` call, reported in + /// every RESET_STREAM_AT frame for the stream. Immutable, even as the reliable size shrinks. + pub(super) final_size: u64, + /// The application error code carried by the reset. Immutable across frames. + pub(super) error_code: VarInt, + /// Whether the most recently transmitted RESET_STREAM_AT frame (carrying the current reliable + /// size) has been acknowledged. Reset to `false` whenever the reliable size is reduced. + pub(super) frame_acked: bool, } impl Send { @@ -33,6 +52,7 @@ impl Send { fin_pending: false, connection_blocked: false, stop_reason: None, + reset_at: None, }) } @@ -55,24 +75,61 @@ impl Send { } } - pub(super) fn reset_at(&mut self, offset: VarInt) -> Result<(), ResetStreamAtError> { - if let Some(error_code) = self.stop_reason { - return Err(ResetStreamAtError::Stopped(error_code)); + /// Begin or tighten a reliable reset (RESET_STREAM_AT), committing to delivering stream data up + /// to `reliable_size` before the stream is reset with `error_code`. + /// + /// The send buffer is truncated to the reliable size so data beyond it is no longer + /// (re)transmitted. The committed reliable size is read back from the buffer (it cannot drop + /// below already-acknowledged data). Returns whether a RESET_STREAM_AT frame needs to be + /// (re)queued for transmission. + /// + /// May be called repeatedly to *reduce* the reliable size; increasing it, changing the error + /// code, or calling it after a FIN-based finish or an ordinary reset is rejected. + pub(super) fn reset_at( + &mut self, + reliable_size: VarInt, + error_code: VarInt, + ) -> Result { + if let Some(code) = self.stop_reason { + return Err(ResetStreamAtError::Stopped(code)); } + let reliable_size = reliable_size.into_inner(); match self.state { - SendState::Ready | SendState::DataSent { .. } if offset.0 >= self.pending.offset() => { - Err(ResetStreamAtError::InvalidReliableSize) - } SendState::Ready => { + // The reliable size cannot exceed the data the application has written so far. + if reliable_size > self.pending.offset() { + return Err(ResetStreamAtError::InvalidReliableSize); + } self.state = SendState::DataSent { finish_acked: false, }; - self.pending.truncate(offset.0); - Ok(()) + self.reset_at = Some(ResetAt { + final_size: self.pending.offset(), + error_code, + frame_acked: false, + }); + self.pending.truncate(reliable_size); + Ok(true) } SendState::DataSent { .. } => { - self.pending.truncate(offset.0); - Ok(()) + let current_reliable = self.pending.offset(); + let Some(reset_at) = self.reset_at.as_mut() else { + // The stream was finished with a FIN; reliable reset after FIN is unsupported. + return Err(ResetStreamAtError::ClosedStream); + }; + // The error code and final size are immutable; the reliable size may only shrink. + if error_code != reset_at.error_code || reliable_size > current_reliable { + return Err(ResetStreamAtError::InvalidReliableSize); + } + self.pending.truncate(reliable_size); + if self.pending.offset() < current_reliable { + // The reliable size genuinely shrank, so a new RESET_STREAM_AT carrying it must + // be transmitted and acknowledged afresh. + reset_at.frame_acked = false; + Ok(true) + } else { + Ok(false) + } } SendState::ResetSent => Err(ResetStreamAtError::ClosedStream), } @@ -133,20 +190,45 @@ impl Send { } } - /// Returns whether the stream has been finished and all data has been acknowledged by the peer + /// Returns whether the stream is fully closed and all data has been acknowledged by the peer + /// + /// For a FIN-based finish this means the FIN and all data were acknowledged; for a reliable + /// reset it means the RESET_STREAM_AT frame and all data up to the reliable size were + /// acknowledged. pub(super) fn ack(&mut self, frame: frame::StreamMeta) -> bool { self.pending.ack(frame.offsets); match self.state { SendState::DataSent { ref mut finish_acked, } => { - *finish_acked |= frame.fin; - *finish_acked && self.pending.is_fully_acked() + if let Some(reset_at) = &self.reset_at { + // A reliable reset completes once the RESET_STREAM_AT frame and all data up to + // the reliable size have been acknowledged. The FIN bit is never set on these + // streams, so `finish_acked` is irrelevant here. + reset_at.frame_acked && self.pending.is_fully_acked() + } else { + *finish_acked |= frame.fin; + *finish_acked && self.pending.is_fully_acked() + } } _ => false, } } + /// Records acknowledgement of a RESET_STREAM_AT frame carrying the current reliable size. + /// + /// Returns whether the reliable reset is now complete (the frame and all reliable data have + /// been acknowledged), in which case the stream may be freed. + pub(super) fn reset_at_acked(&mut self) -> bool { + match &mut self.reset_at { + Some(reset_at) => { + reset_at.frame_acked = true; + self.pending.is_fully_acked() + } + None => false, + } + } + /// Handle increase to stream-level flow control limit /// /// Returns whether the stream was unblocked @@ -345,6 +427,131 @@ pub enum FinishError { mod tests { use super::*; + /// A `Send` with `data` bytes written and ready to be reset. + fn writer(data: &[u8]) -> Box { + let mut send = Send::new(VarInt::MAX); + let mut source = ByteSlice::from_slice(data); + send.write(&mut source, data.len() as u64).unwrap(); + assert_eq!(send.offset(), data.len() as u64); + send + } + + #[test] + fn reset_at_from_ready_truncates_and_keeps_final_size() { + let mut send = writer(b"0123456789"); // 10 bytes + assert!(send.reset_at(4u32.into(), 7u32.into()).unwrap()); + + // The buffer is truncated to the reliable size, but the final size is remembered. + assert_eq!( + send.offset(), + 4, + "reliable size becomes the send-buffer end" + ); + let reset_at = send.reset_at.as_ref().expect("reliable reset in progress"); + assert_eq!(reset_at.final_size, 10); + assert_eq!(reset_at.error_code, VarInt::from_u32(7)); + assert!(!reset_at.frame_acked); + // The stream is closing: not writable, but not a full reset either (data still flows). + assert!(!send.is_writable()); + assert!(!send.is_reset()); + } + + #[test] + fn reset_at_reliable_size_may_equal_final_size() { + let mut send = writer(b"0123456789"); + // Reliable size == bytes written is valid (deliver everything, then signal the reset). + assert!(send.reset_at(10u32.into(), 0u32.into()).unwrap()); + assert_eq!(send.offset(), 10); + assert_eq!(send.reset_at.as_ref().unwrap().final_size, 10); + } + + #[test] + fn reset_at_reliable_size_beyond_written_is_rejected() { + let mut send = writer(b"0123456789"); + assert_eq!( + send.reset_at(11u32.into(), 0u32.into()), + Err(ResetStreamAtError::InvalidReliableSize) + ); + // The stream is untouched and still writable. + assert!(send.is_writable()); + assert!(send.reset_at.is_none()); + } + + #[test] + fn reset_at_on_stopped_stream_is_rejected() { + let mut send = writer(b"0123456789"); + assert!(send.try_stop(9u32.into())); + assert_eq!( + send.reset_at(4u32.into(), 0u32.into()), + Err(ResetStreamAtError::Stopped(9u32.into())) + ); + } + + #[test] + fn reset_at_after_full_reset_is_rejected() { + let mut send = writer(b"0123456789"); + send.reset(); + assert_eq!( + send.reset_at(4u32.into(), 0u32.into()), + Err(ResetStreamAtError::ClosedStream) + ); + } + + #[test] + fn reset_at_may_only_reduce_reliable_size() { + let mut send = writer(b"0123456789"); + assert!(send.reset_at(6u32.into(), 7u32.into()).unwrap()); + + // Reducing genuinely shrinks the buffer and requires re-acknowledging the frame. + assert!(send.reset_at(3u32.into(), 7u32.into()).unwrap()); + assert_eq!(send.offset(), 3); + assert!(!send.reset_at.as_ref().unwrap().frame_acked); + + // Re-requesting the same size is a no-op (no new frame needed). + assert!(!send.reset_at(3u32.into(), 7u32.into()).unwrap()); + assert_eq!(send.offset(), 3); + + // Increasing the reliable size, or changing the error code, is rejected. + assert_eq!( + send.reset_at(5u32.into(), 7u32.into()), + Err(ResetStreamAtError::InvalidReliableSize) + ); + assert_eq!( + send.reset_at(2u32.into(), 8u32.into()), + Err(ResetStreamAtError::InvalidReliableSize) + ); + // The final size never changes across reductions. + assert_eq!(send.reset_at.as_ref().unwrap().final_size, 10); + } + + #[test] + fn reset_at_completes_only_when_frame_and_data_acknowledged() { + // Data acknowledged first, then the frame. + let mut send = writer(b"0123456789"); + assert!(send.reset_at(4u32.into(), 0u32.into()).unwrap()); + let meta = frame::StreamMeta { + id: crate::StreamId::new(crate::Side::Client, crate::Dir::Uni, 0), + offsets: 0..4, + fin: false, + }; + assert!(!send.ack(meta), "data acked but RESET_STREAM_AT not yet"); + assert!( + send.reset_at_acked(), + "frame ack now completes the reliable reset" + ); + + // Frame acknowledged first, then the data. + let mut send = writer(b"0123456789"); + assert!(send.reset_at(4u32.into(), 0u32.into()).unwrap()); + assert!(!send.reset_at_acked(), "frame acked but data not yet"); + let meta = frame::StreamMeta { + id: crate::StreamId::new(crate::Side::Client, crate::Dir::Uni, 0), + offsets: 0..4, + fin: false, + }; + assert!(send.ack(meta), "data ack now completes the reliable reset"); + } + #[test] fn bytes_array() { let full = b"Hello World 123456789 ABCDEFGHJIJKLMNOPQRSTUVWXYZ".to_owned(); diff --git a/noq-proto/src/connection/streams/state.rs b/noq-proto/src/connection/streams/state.rs index 79fb6773c3..c810b1826d 100644 --- a/noq-proto/src/connection/streams/state.rs +++ b/noq-proto/src/connection/streams/state.rs @@ -8,8 +8,8 @@ use rustc_hash::FxHashMap; use tracing::{debug, trace}; use super::{ - PendingStreamsQueue, Recv, Retransmits, Send, SendState, ShouldTransmit, StreamEvent, - StreamHalf, + PendingStreamsQueue, Recv, ResetAtOutcome, Retransmits, Send, SendState, ShouldTransmit, + StreamEvent, StreamHalf, }; use crate::{ Dir, MAX_STREAM_COUNT, Side, StreamId, TransportError, VarInt, @@ -137,6 +137,9 @@ pub struct StreamsState { receive_window_shrink_debt: u64, /// Whether the locally-initiated stream limit has been hit, per direction pub(super) streams_blocked: [bool; 2], + /// Whether the peer advertised the `reset_stream_at` transport parameter, i.e. whether it can + /// receive RESET_STREAM_AT frames. Gates use of [`SendStream::reset_at`]. + pub(super) peer_reset_stream_at: bool, } impl StreamsState { @@ -182,10 +185,12 @@ impl StreamsState { initial_max_stream_data_bidi_remote: 0u32.into(), receive_window_shrink_debt: 0, streams_blocked: [false, false], + peer_reset_stream_at: false, } } pub(crate) fn set_params(&mut self, params: &TransportParameters) { + self.peer_reset_stream_at = params.reset_stream_at; self.initial_max_stream_data_uni = params.initial_max_stream_data_uni; self.initial_max_stream_data_bidi_local = params.initial_max_stream_data_bidi_local; self.initial_max_stream_data_bidi_remote = params.initial_max_stream_data_bidi_remote; @@ -320,6 +325,12 @@ impl StreamsState { return Ok(ShouldTransmit(false)); }; + // A plain RESET_STREAM following a RESET_STREAM_AT reduces the reliable size to 0. The tail + // beyond the reliable size already had its connection credit released (and `end` was + // advanced to the final size) when the RESET_STREAM_AT was processed, so capture the + // still-deliverable region now to avoid re-issuing that credit below. + let prior_deliver_cap = rs.reliable_reset_deliver_cap(); + // State transition if !rs.reset( error_code, @@ -344,7 +355,11 @@ impl StreamsState { } // Update connection-level flow control - Ok(if bytes_read != final_offset.into_inner() { + Ok(if let Some(deliver_cap) = prior_deliver_cap { + // Downgrade from a reliable reset: release only the still-deliverable region; the rest + // of the final size was already accounted when the RESET_STREAM_AT arrived. + self.add_read_credits(deliver_cap.saturating_sub(bytes_read)) + } else if bytes_read != final_offset.into_inner() { // bytes_read is always <= end, so this won't underflow. self.data_recvd = self .data_recvd @@ -355,6 +370,90 @@ impl StreamsState { }) } + /// Process incoming RESET_STREAM_AT (reliable reset) frame + /// + /// If successful, returns whether a `MAX_DATA` frame needs to be transmitted. + #[allow(unreachable_pub)] // fuzzing only + pub fn received_reset_at( + &mut self, + frame: frame::ResetStreamAt, + ) -> Result { + let frame::ResetStreamAt { + id, + error_code, + final_offset, + reliable_size, + } = frame; + self.validate_receive_id(id).inspect_err(|_e| { + debug!("received illegal RESET_STREAM_AT frame"); + })?; + + // Create state for this stream if the remote peer created it. + let newly_created = self.ensure_remote(id); + + let Some(rs) = self + .recv + .get_mut(&id) + .map(get_or_insert_recv(self.stream_receive_window)) + else { + trace!("received RESET_STREAM_AT on closed stream"); + return Ok(ShouldTransmit(false)); + }; + + // A stopped stream discards incoming data, so partial delivery is moot: treat the reliable + // reset exactly like an ordinary RESET_STREAM, accounting the final size and disposing of + // the stream immediately. + if rs.stopped { + if !rs.reset( + error_code, + final_offset, + self.data_recvd, + self.local_max_data, + )? { + // Redundant reset + return Ok(ShouldTransmit(false)); + } + let bytes_read = rs.assembler.bytes_read(); + let end = rs.end; + let rs = self.recv.remove(&id).flatten().unwrap(); + self.stream_recv_freed(id, rs); + return Ok(if bytes_read != final_offset.into_inner() { + self.data_recvd = self + .data_recvd + .saturating_add(u64::from(final_offset) - end); + self.add_read_credits(u64::from(final_offset) - bytes_read) + } else { + ShouldTransmit(false) + }); + } + + let outcome = rs.reset_at( + error_code, + final_offset, + reliable_size, + self.data_recvd, + self.local_max_data, + )?; + + let ResetAtOutcome::Applied { + received_delta, + credit, + } = outcome + else { + // Redundant, or an attempt to increase the reliable size, which is ignored. + return Ok(ShouldTransmit(false)); + }; + + if !newly_created { + // Newly opened streams are inherently readable; the app discovers them via + // `StreamEvent::Opened` and a separate `Readable` would be redundant. + self.events.push_back(StreamEvent::Readable { id }); + } + + self.data_recvd = self.data_recvd.saturating_add(received_delta); + Ok(self.add_read_credits(credit)) + } + /// Process incoming `STOP_SENDING` frame #[allow(unreachable_pub)] // fuzzing only pub fn received_stop_sending(&mut self, id: StreamId, error_code: VarInt) { @@ -388,6 +487,34 @@ impl StreamsState { } } + /// Process acknowledgement of a RESET_STREAM_AT frame carrying `reliable_size`. + /// + /// If this completes the reliable reset (the frame and all data up to the reliable size are + /// acknowledged), the send half is freed and a [`StreamEvent::Finished`] is emitted, just as + /// for a FIN-based finish. The completion may instead be observed via the stream-data ack path + /// ([`Self::received_ack_of`]); whichever arrives last frees the stream. + /// + /// Acknowledgements of a frame whose reliable size has since been reduced are ignored, so that + /// the smallest reliable size is retransmitted until it is itself acknowledged. + pub(crate) fn reset_at_acked(&mut self, id: StreamId, reliable_size: VarInt) { + let hash_map::Entry::Occupied(mut e) = self.send.entry(id) else { + return; + }; + let Some(stream) = e.get_mut().as_mut() else { + return; + }; + // The current (smallest) reliable size equals the truncated send-buffer end. A larger value + // belongs to a superseded frame whose acknowledgement must not complete the reset. + if reliable_size.into_inner() != stream.pending.offset() { + return; + } + if stream.reset_at_acked() { + e.remove_entry(); + self.stream_freed(id, StreamHalf::Send); + self.events.push_back(StreamEvent::Finished { id }); + } + } + /// Whether any stream data is queued, regardless of control frames pub(crate) fn can_send_stream_data(&self) -> bool { // Reset streams may linger in the pending stream list, but will never produce stream frames @@ -430,6 +557,33 @@ impl StreamsState { builder.write_frame(frame, stats); } + // RESET_STREAM_AT + while builder.frame_space_remaining() > frame::ResetStreamAt::SIZE_BOUND { + // The stored reliable size is only used for acknowledgement tracking; the frame is + // always (re)built from the stream's current (smallest) reliable size below. + let Some((id, _)) = pending.reset_stream_at.pop() else { + break; + }; + let Some(stream) = self.send.get_mut(&id).and_then(|s| s.as_mut()) else { + continue; + }; + // The reliable reset may have been superseded or already completed and freed. + let Some(reset_at) = stream.reset_at.as_ref() else { + continue; + }; + // The reliable size is the current send-buffer end (the buffer was truncated to it); + // the final size and error code are fixed for the lifetime of the reliable reset. + let frame = frame::ResetStreamAt { + id, + error_code: reset_at.error_code, + final_offset: VarInt::try_from(reset_at.final_size) + .expect("impossibly large offset"), + reliable_size: VarInt::try_from(stream.pending.offset()) + .expect("impossibly large offset"), + }; + builder.write_frame(frame, stats); + } + // STOP_SENDING while builder.frame_space_remaining() > frame::StopSending::SIZE_BOUND { let Some(frame) = pending.stop_sending.pop() else { @@ -549,8 +703,11 @@ impl StreamsState { // are required to encode it. let max_buf_size = builder.frame_space_remaining() - 1 - VarInt::size(id.into()); let (offsets, encode_length) = stream.pending.poll_transmit(max_buf_size); + // A reliable reset (RESET_STREAM_AT) signals the end of the stream itself, so its data + // frames must not also carry a FIN, which would imply a (smaller) clean final size. let fin = offsets.end == stream.pending.offset() - && matches!(stream.state, SendState::DataSent { .. }); + && matches!(stream.state, SendState::DataSent { .. }) + && stream.reset_at.is_none(); if fin { stream.fin_pending = false; } @@ -585,7 +742,7 @@ impl StreamsState { builder.sent_frames().stream_frames.clone() } - pub(crate) fn received_ack_of(&mut self, frame: frame::StreamMeta) { + pub(crate) fn received_ack_of(&mut self, mut frame: frame::StreamMeta) { let mut entry = match self.send.entry(frame.id) { hash_map::Entry::Vacant(_) => return, hash_map::Entry::Occupied(e) => e, @@ -604,6 +761,15 @@ impl StreamsState { return; } let id = frame.id; + if stream.reset_at.is_some() { + // A reliable reset truncated the send buffer to the reliable size and restored the + // window credit for the abandoned tail at reset time. Acknowledgements of that tail's + // in-flight data must therefore be ignored here, both for window accounting and to + // avoid feeding the send buffer a range it no longer holds. + let reliable_size = stream.pending.offset(); + frame.offsets.end = frame.offsets.end.min(reliable_size); + frame.offsets.start = frame.offsets.start.min(frame.offsets.end); + } self.unacked_data -= frame.offsets.end - frame.offsets.start; if !stream.ack(frame) { // The stream is unfinished or may still need retransmits @@ -620,11 +786,20 @@ impl StreamsState { // Loss of data on a closed stream is a noop return; }; + let mut offsets = frame.offsets; + if stream.reset_at.is_some() { + // Data beyond a reliable reset's reliable size was abandoned (the send buffer was + // truncated to it) and must not be retransmitted. + offsets.end = offsets.end.min(stream.pending.offset()); + } + if offsets.start >= offsets.end { + return; + } if !stream.is_pending() { self.pending.push_pending(frame.id, stream.priority); } stream.fin_pending |= frame.fin; - stream.pending.retransmit(frame.offsets); + stream.pending.retransmit(offsets); } pub(crate) fn retransmit_all_for_0rtt(&mut self) { @@ -965,8 +1140,8 @@ pub(super) fn get_or_insert_recv( mod tests { use super::*; use crate::{ - ReadableError, RecvStream, SendStream, TransportErrorCode, WriteError, - connection::State as ConnState, connection::Streams, + ReadableError, ReadError, RecvStream, ResetStreamAtError, SendStream, TransportErrorCode, + WriteError, connection::State as ConnState, connection::Streams, }; use bytes::Bytes; @@ -2227,4 +2402,412 @@ mod tests { assert_eq!(stream.write(&data), Ok(smaller_send_window as usize)); assert_eq!(stream.write(&data), Err(WriteError::Blocked)); } + + /// Reads a reliably-reset stream to completion, returning the delivered byte count and the + /// surfaced reset error code. + fn drain_reliable_reset(client: &mut StreamsState, id: StreamId) -> (usize, VarInt) { + let mut pending = Retransmits::default(); + let mut recv = RecvStream { + id, + state: client, + pending: &mut pending, + }; + let mut chunks = recv.read(true).unwrap(); + let mut delivered = 0; + let code = loop { + match chunks.next(4096) { + Ok(Some(chunk)) => delivered += chunk.bytes.len(), + Ok(None) => panic!("a reliable reset must surface a reset, not a clean finish"), + Err(ReadError::Reset(code)) => break code, + Err(ReadError::Blocked) => panic!("unexpected block while draining"), + } + }; + let _ = chunks.finalize(); + (delivered, code) + } + + #[test] + fn reliable_reset_delivers_prefix_then_reset() { + let mut client = make(Side::Client); + let id = StreamId::new(Side::Server, Dir::Uni, 0); + let initial_max = client.local_max_data; + + // Receive 100 bytes, none read yet. + let _ = client + .received( + frame::Stream { + id, + offset: 0, + fin: false, + data: Bytes::from_static(&[0; 100]), + }, + 100, + ) + .unwrap(); + assert_eq!(client.data_recvd, 100); + assert_eq!(client.local_max_data - initial_max, 0); + + // Reliable reset: final size 100, deliver up to 40, error code 7. + let _ = client + .received_reset_at(frame::ResetStreamAt { + id, + error_code: 7u32.into(), + final_offset: 100u32.into(), + reliable_size: 40u32.into(), + }) + .unwrap(); + // The never-delivered tail [40, 100) releases its connection credit immediately. + assert_eq!(client.data_recvd, 100); + assert_eq!(client.local_max_data - initial_max, 60); + + // The application reads exactly the reliable prefix, then observes the reset. + let (delivered, code) = drain_reliable_reset(&mut client, id); + assert_eq!(delivered, 40); + assert_eq!(code, VarInt::from_u32(7)); + + // Reading the delivered prefix releases the remaining credit: in total, the full final + // size of connection flow control credit is returned. + assert_eq!(client.local_max_data - initial_max, 100); + } + + #[test] + fn reliable_reset_waits_for_retransmit_of_reliable_data() { + let mut client = make(Side::Client); + let id = StreamId::new(Side::Server, Dir::Uni, 0); + + // Receive [10, 100) first, leaving a gap at [0, 10). + let _ = client + .received( + frame::Stream { + id, + offset: 10, + fin: false, + data: Bytes::from_static(&[0; 90]), + }, + 90, + ) + .unwrap(); + let _ = client + .received_reset_at(frame::ResetStreamAt { + id, + error_code: 5u32.into(), + final_offset: 100u32.into(), + reliable_size: 40u32.into(), + }) + .unwrap(); + + // Reading blocks: the reliable prefix has a hole that must be retransmitted. + let mut pending = Retransmits::default(); + { + let mut recv = RecvStream { + id, + state: &mut client, + pending: &mut pending, + }; + let mut chunks = recv.read(true).unwrap(); + assert_eq!(chunks.next(4096), Err(ReadError::Blocked)); + let _ = chunks.finalize(); + } + + // The sender retransmits the missing reliable data; the stream must still accept it. + let _ = client + .received( + frame::Stream { + id, + offset: 0, + fin: false, + data: Bytes::from_static(&[0; 10]), + }, + 10, + ) + .unwrap(); + + let (delivered, code) = drain_reliable_reset(&mut client, id); + assert_eq!(delivered, 40); + assert_eq!(code, VarInt::from_u32(5)); + } + + #[test] + fn reliable_reset_on_stopped_stream_is_a_plain_reset() { + 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(&[0; 50]), + }, + 50, + ) + .unwrap(); + + // Stop the stream: it discards data, so partial delivery is moot. + let mut pending = Retransmits::default(); + { + let mut recv = RecvStream { + id, + state: &mut client, + pending: &mut pending, + }; + recv.stop(0u32.into()).unwrap(); + } + + // A reliable reset on a stopped stream is handled exactly like an ordinary RESET_STREAM: + // the final size is accounted for connection flow control and the stream is freed + // immediately, with no partial delivery. + let _ = client + .received_reset_at(frame::ResetStreamAt { + id, + error_code: 1u32.into(), + final_offset: 80u32.into(), + reliable_size: 40u32.into(), + }) + .unwrap(); + assert_eq!( + client.data_recvd, 80, + "final size accounted for flow control" + ); + assert!( + !client.recv.contains_key(&id), + "stopped stream is freed on reset" + ); + } + + #[test] + fn reset_stream_after_reliable_reset_does_not_over_issue_credit() { + let mut client = make(Side::Client); + let id = StreamId::new(Side::Server, Dir::Uni, 0); + let initial_max = client.local_max_data; + + let _ = client + .received( + frame::Stream { + id, + offset: 0, + fin: false, + data: Bytes::from_static(&[0; 100]), + }, + 100, + ) + .unwrap(); + + // Reliable reset (final 100, reliable 40) releases the [40, 100) tail credit immediately. + let _ = client + .received_reset_at(frame::ResetStreamAt { + id, + error_code: 7u32.into(), + final_offset: 100u32.into(), + reliable_size: 40u32.into(), + }) + .unwrap(); + assert_eq!(client.local_max_data - initial_max, 60); + + // A plain RESET_STREAM follows (the peer reduces the reliable size to 0). It must release + // only the still-deliverable [0, 40) credit, not re-issue the tail already released above; + // total credit over the stream lifetime must equal the final size, never exceed it. + let _ = client + .received_reset(frame::ResetStream { + id, + error_code: 7u32.into(), + final_offset: 100u32.into(), + }) + .unwrap(); + assert_eq!( + client.local_max_data - initial_max, + 100, + "total credit must not exceed the final size" + ); + } + + #[test] + fn reliable_reset_unordered_read_delivers_prefix() { + let mut client = make(Side::Client); + let id = StreamId::new(Side::Server, Dir::Uni, 0); + + // Receive [60, 100) (beyond the reliable size) while [0, 40) is still missing. + let _ = client + .received( + frame::Stream { + id, + offset: 60, + fin: false, + data: Bytes::from_static(&[0; 40]), + }, + 40, + ) + .unwrap(); + let _ = client + .received_reset_at(frame::ResetStreamAt { + id, + error_code: 9u32.into(), + final_offset: 100u32.into(), + reliable_size: 40u32.into(), + }) + .unwrap(); + + // An unordered read must not hand out the buffered post-reliable-size data, nor surface the + // reset early: it blocks waiting for the reliable prefix. + let mut pending = Retransmits::default(); + { + let mut recv = RecvStream { + id, + state: &mut client, + pending: &mut pending, + }; + let mut chunks = recv.read(false).unwrap(); + assert_eq!(chunks.next(4096), Err(ReadError::Blocked)); + let _ = chunks.finalize(); + } + + // The reliable prefix arrives; an unordered read now delivers exactly [0, 40), then the + // reset. The buffered [60, 100) is never delivered. + let _ = client + .received( + frame::Stream { + id, + offset: 0, + fin: false, + data: Bytes::from_static(&[0; 40]), + }, + 40, + ) + .unwrap(); + + let mut pending = Retransmits::default(); + let mut recv = RecvStream { + id, + state: &mut client, + pending: &mut pending, + }; + let mut chunks = recv.read(false).unwrap(); + let mut delivered = 0; + let code = loop { + match chunks.next(4096) { + Ok(Some(chunk)) => delivered += chunk.bytes.len(), + Ok(None) => panic!("a reliable reset must surface a reset"), + Err(ReadError::Reset(code)) => break code, + Err(ReadError::Blocked) => panic!("unexpected block after reliable prefix arrived"), + } + }; + let _ = chunks.finalize(); + assert_eq!( + delivered, 40, + "delivers the reliable prefix, not the buffered tail" + ); + assert_eq!(code, VarInt::from_u32(9)); + } + + /// Opens a server-initiated uni send stream with the peer advertising `reset_stream_at`. + fn send_stream_setup(peer_supports_reliable_reset: bool) -> (StreamsState, StreamId) { + let mut server = make(Side::Server); + server.set_params(&TransportParameters { + initial_max_streams_uni: 1u32.into(), + initial_max_data: 1024u32.into(), + initial_max_stream_data_uni: 1024u32.into(), + reset_stream_at: peer_supports_reliable_reset, + ..TransportParameters::default() + }); + let id = { + let state = ConnState::established(); + let mut streams = Streams { + state: &mut server, + conn_state: &state, + }; + streams.open(Dir::Uni).unwrap() + }; + (server, id) + } + + #[test] + fn reliable_reset_send_emits_prefix_without_fin_and_queues_frame() { + let (mut server, id) = send_stream_setup(true); + let state = ConnState::established(); + let mut pending = Retransmits::default(); + { + let mut stream = SendStream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &state, + }; + stream.write(b"0123456789").unwrap(); // 10 bytes + stream.reset_at(4u32.into(), 7u32.into()).unwrap(); + } + + // The stream is queued for a RESET_STREAM_AT frame carrying the reliable size, and still + // has reliable data to send. + assert_eq!(pending.reset_stream_at, &[(id, VarInt::from_u32(4))]); + assert!(server.can_send_stream_data()); + + // Only the reliable prefix [0, 4) is sent, and never with a FIN (the frame ends the stream). + let metas = server.write_frames_for_test(1200, true); + let sent: u64 = metas.iter().map(|m| m.offsets.end - m.offsets.start).sum(); + assert_eq!(sent, 4, "only the reliable prefix is transmitted"); + assert!( + metas.iter().all(|m| !m.fin), + "RESET_STREAM_AT replaces the FIN bit" + ); + } + + #[test] + fn reset_at_requires_peer_support() { + let (mut server, id) = send_stream_setup(false); + let state = ConnState::established(); + let mut pending = Retransmits::default(); + let mut stream = SendStream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &state, + }; + stream.write(b"hello").unwrap(); + assert_eq!( + stream.reset_at(2u32.into(), 0u32.into()), + Err(ResetStreamAtError::Unsupported), + ); + } + + #[test] + fn reduced_reliable_reset_ignores_stale_frame_ack() { + let (mut server, id) = send_stream_setup(true); + let state = ConnState::established(); + let mut pending = Retransmits::default(); + { + let mut stream = SendStream { + id, + state: &mut server, + pending: &mut pending, + conn_state: &state, + }; + stream.write(b"0123456789").unwrap(); + stream.reset_at(8u32.into(), 7u32.into()).unwrap(); // reliable size 8 + stream.reset_at(4u32.into(), 7u32.into()).unwrap(); // reduced to 4 + } + + // All data up to the (reduced) reliable size is acknowledged. + server.received_ack_of(frame::StreamMeta { + id, + offsets: 0..4, + fin: false, + }); + assert!(server.send.contains_key(&id), "frame not yet acknowledged"); + + // An acknowledgement of the superseded frame (reliable size 8) must not complete the reset: + // the smallest reliable size must keep being retransmitted until it is itself acknowledged. + server.reset_at_acked(id, VarInt::from_u32(8)); + assert!( + server.send.contains_key(&id), + "stale frame ack must not free the stream" + ); + + // Acknowledging the current frame completes the reliable reset. + server.reset_at_acked(id, VarInt::from_u32(4)); + assert!( + !server.send.contains_key(&id), + "current frame ack completes the reliable reset" + ); + } } diff --git a/noq-proto/src/frame.rs b/noq-proto/src/frame.rs index f8a06e7b02..b9da67edea 100644 --- a/noq-proto/src/frame.rs +++ b/noq-proto/src/frame.rs @@ -2517,14 +2517,28 @@ impl Encodable for RemoveAddress { #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] #[cfg_attr(test, derive(Arbitrary))] #[derive(Debug, Copy, Clone, derive_more::Display)] -#[display("RESET_STREAM id: {id}")] +#[display("RESET_STREAM_AT id: {id}, final: {final_offset}, reliable: {reliable_size}")] pub(crate) struct ResetStreamAt { pub(crate) id: StreamId, pub(crate) error_code: VarInt, pub(crate) final_offset: VarInt, + // The wire format requires `reliable_size <= final_offset`; a frame violating this is a + // `FRAME_ENCODING_ERROR` on decode. Constrain generated values so that the encode/decode + // round-trip proptest only produces decodable frames. + #[cfg_attr(test, strategy(reset_stream_at_reliable_size(#final_offset)))] pub(crate) reliable_size: VarInt, } +/// Proptest strategy producing a `reliable_size` no larger than `final_offset`, as required by +/// the RESET_STREAM_AT wire format. +#[cfg(test)] +fn reset_stream_at_reliable_size( + final_offset: VarInt, +) -> impl proptest::strategy::Strategy { + use proptest::strategy::Strategy; + (0..=final_offset.0).prop_map(VarInt) +} + impl ResetStreamAt { pub(crate) const fn get_type(&self) -> FrameType { FrameType::ResetStreamAt @@ -2537,7 +2551,7 @@ impl FrameStruct for ResetStreamAt { impl Encodable for ResetStreamAt { fn encode(&self, out: &mut W) { - out.write(FrameType::ResetStream); // 1 byte + out.write(FrameType::ResetStreamAt); // 1 byte out.write(self.id); // <= 8 bytes out.write(self.error_code); // <= 8 bytes out.write(self.final_offset); // <= 8 bytes diff --git a/noq-proto/src/tests/encode_decode.rs b/noq-proto/src/tests/encode_decode.rs index f822be9085..9a293481d0 100644 --- a/noq-proto/src/tests/encode_decode.rs +++ b/noq-proto/src/tests/encode_decode.rs @@ -92,3 +92,54 @@ fn maybe_frame_known_never_padding(frame: MaybeFrame) { prop_assert_ne!(ft, FrameType::Padding); } } + +#[test] +fn reset_stream_at_type_byte_and_roundtrip() { + use crate::frame::ResetStreamAt; + use crate::{Dir, Side, StreamId, VarInt}; + + let frame = ResetStreamAt { + id: StreamId::new(Side::Client, Dir::Uni, 3), + error_code: VarInt::from_u32(42), + final_offset: VarInt::from_u32(1000), + reliable_size: VarInt::from_u32(250), + }; + + let mut encoded = BytesMut::new(); + encode_frame(&Frame::ResetStreamAt(frame), &mut encoded); + // The RFC assigns RESET_STREAM_AT the frame type 0x24, which fits in a single varint byte. + assert_eq!(encoded[0], 0x24, "RESET_STREAM_AT must use frame type 0x24"); + + let mut iter = crate::frame::Iter::new(encoded.freeze()).unwrap(); + let decoded = iter.next().unwrap().unwrap(); + assert_eq!(decoded, Frame::ResetStreamAt(frame)); + assert!(iter.take_remaining().is_empty()); +} + +#[test] +fn reset_stream_at_reliable_exceeds_final_is_rejected() { + use crate::TransportErrorCode; + use crate::frame::ResetStreamAt; + use crate::{Dir, Side, StreamId, VarInt}; + + // A reliable size larger than the final size is malformed and must be rejected with a + // FRAME_ENCODING_ERROR (draft-ietf-quic-reliable-stream-reset section 4). `encode` does not + // validate, so we can construct the illegal wire bytes directly. + let frame = ResetStreamAt { + id: StreamId::new(Side::Client, Dir::Bi, 0), + error_code: VarInt::from_u32(7), + final_offset: VarInt::from_u32(10), + reliable_size: VarInt::from_u32(11), + }; + + let mut encoded = BytesMut::new(); + encode_frame(&Frame::ResetStreamAt(frame), &mut encoded); + + let mut iter = crate::frame::Iter::new(encoded.freeze()).unwrap(); + let err = iter + .next() + .unwrap() + .expect_err("reliable > final must not decode"); + let err = crate::TransportError::from(err); + assert_eq!(err.code, TransportErrorCode::FRAME_ENCODING_ERROR); +} diff --git a/noq-proto/src/transport_parameters.rs b/noq-proto/src/transport_parameters.rs index b721c2071c..e86438db32 100644 --- a/noq-proto/src/transport_parameters.rs +++ b/noq-proto/src/transport_parameters.rs @@ -221,7 +221,9 @@ impl TransportParameters { || cached.grease_quic_bit && !self.grease_quic_bit || cached.address_discovery_role != self.address_discovery_role || cached.max_remote_nat_traversal_addresses != self.max_remote_nat_traversal_addresses - || cached.reset_stream_at != self.reset_stream_at + // The extension may be newly enabled on resumption, but a server MUST NOT disable it + // (draft-ietf-quic-reliable-stream-reset §3). + || cached.reset_stream_at && !self.reset_stream_at { return Err(TransportError::PROTOCOL_VIOLATION( "0-RTT accepted with incompatible transport parameters", @@ -1001,4 +1003,27 @@ mod test { high_limit.validate_resumption_from(&low_limit).unwrap(); low_limit.validate_resumption_from(&high_limit).unwrap_err(); } + + #[test] + fn reset_stream_at_empty_value_enables_support() { + // The parameter is advertised with an empty value. + let mut buf = Vec::new(); + buf.write_var(TransportParameterId::ResetStreamAt as u64); + buf.write_var(0); + let params = TransportParameters::read(Side::Server, &mut buf.as_slice()).unwrap(); + assert!(params.reset_stream_at); + } + + #[test] + fn reset_stream_at_rejects_non_empty_value() { + // A non-empty `reset_stream_at` value must be rejected. + let mut buf = Vec::new(); + buf.write_var(TransportParameterId::ResetStreamAt as u64); + buf.write_var(1); + buf.put_u8(0); + assert_eq!( + TransportParameters::read(Side::Server, &mut buf.as_slice()), + Err(Error::Malformed) + ); + } } diff --git a/noq/src/lib.rs b/noq/src/lib.rs index ac43d2ee69..c570437635 100644 --- a/noq/src/lib.rs +++ b/noq/src/lib.rs @@ -66,10 +66,10 @@ pub use proto::{ ConnectionIdGenerator, ConnectionStats, DecryptedInitial, Dir, EcnCodepoint, EndpointConfig, FourTuple, FrameStats, FrameType, IdleTimeout, InvalidCid, MtuDiscoveryConfig, NetworkChangeHint, NoneTokenLog, NoneTokenStore, PathError, PathEvent, PathId, PathStats, - PathStatus, ServerConfig, SetPathStatusError, Side, StdSystemTime, StreamId, TimeSource, - TokenLog, TokenMemoryCache, TokenReuseError, TokenStore, Transmit, TransportConfig, - TransportErrorCode, UdpStats, ValidationTokenConfig, VarInt, VarIntBoundsExceeded, congestion, - crypto, + PathStatus, ResetStreamAtError, ServerConfig, SetPathStatusError, Side, StdSystemTime, + StreamId, TimeSource, TokenLog, TokenMemoryCache, TokenReuseError, TokenStore, Transmit, + TransportConfig, TransportErrorCode, UdpStats, ValidationTokenConfig, VarInt, + VarIntBoundsExceeded, congestion, crypto, }; #[cfg(feature = "qlog")] pub use proto::{QlogConfig, QlogFactory, QlogFileFactory}; diff --git a/noq/src/send_stream.rs b/noq/src/send_stream.rs index 534e32d853..724758e3dc 100644 --- a/noq/src/send_stream.rs +++ b/noq/src/send_stream.rs @@ -219,16 +219,26 @@ impl SendStream { Ok(()) } - /// Close the send stream immediately, delivering data up to `offset`. + /// Close the send stream, reliably delivering data up to `reliable_size` before the reset. /// - /// No new data can be written after calling this method. Data up to `offset` is still - /// reliably delivered, but any remaining stream data may never be transmitted or lost. + /// Sends a RESET_STREAM_AT frame ([draft-ietf-quic-reliable-stream-reset]): the peer is + /// delivered all stream data up to `reliable_size` and only then observes the reset carrying + /// `error_code`. Unlike [`Self::reset`], data up to the reliable size is retransmitted on loss. + /// No new data can be written after calling this; any data beyond the reliable size may never + /// be delivered. /// - /// Fails if [`Self::finish`], [`Self::reset`] or [`Self::reset_at`] was previously - /// called, of if the remote stopped the stream. + /// May be called repeatedly to *reduce* `reliable_size` (the error code must stay the same). + /// `reliable_size` may not exceed the number of bytes written to the stream. + /// + /// Fails with [`ResetStreamAtError::Unsupported`] if the peer did not advertise support for the + /// extension, or with [`ResetStreamAtError::ClosedStream`] if [`Self::finish`] or + /// [`Self::reset`] was previously called, or [`ResetStreamAtError::Stopped`] if the peer + /// stopped the stream. + /// + /// [draft-ietf-quic-reliable-stream-reset]: https://datatracker.ietf.org/doc/html/draft-ietf-quic-reliable-stream-reset pub fn reset_at( &mut self, - offset: VarInt, + reliable_size: VarInt, error_code: VarInt, ) -> Result<(), ResetStreamAtError> { let mut conn = self.conn.lock_and_wake("SendStream::reset_at"); @@ -238,7 +248,7 @@ impl SendStream { } conn.inner .send_stream(self.stream) - .reset_at(offset, error_code)?; + .reset_at(reliable_size, error_code)?; Ok(()) } diff --git a/noq/src/tests.rs b/noq/src/tests.rs index 74c760c3de..ecc6978dcc 100755 --- a/noq/src/tests.rs +++ b/noq/src/tests.rs @@ -181,6 +181,60 @@ fn read_after_close() { }); } +/// End-to-end check of RESET_STREAM_AT: the receiver is delivered the reliable prefix of a stream +/// and then observes the reset with its application error code. +#[tokio::test] +async fn reliable_stream_reset_delivers_prefix() { + let _guard = subscribe(); + let endpoint = endpoint(); + + const MSG: &[u8] = b"the quick brown fox jumps over the lazy dog"; + // "the quick brown fox" -- the prefix the sender commits to delivering reliably. + const RELIABLE: usize = 19; + const CODE: u32 = 7; + + let endpoint2 = endpoint.clone(); + let server = tokio::spawn(async move { + let conn = endpoint2.accept().await.unwrap().await.unwrap(); + let mut s = conn.open_uni().await.unwrap(); + s.write_all(MSG).await.unwrap(); + s.reset_at( + crate::VarInt::from_u32(RELIABLE as u32), + crate::VarInt::from_u32(CODE), + ) + .expect("peer negotiated reset_stream_at"); + // Keep the connection alive until the reliable reset is fully delivered and acknowledged, + // at which point the send half is freed and `stopped()` resolves. + _ = s.stopped().await; + }); + + let conn = endpoint + .connect(endpoint.local_addr().unwrap(), "localhost") + .unwrap() + .await + .unwrap(); + let mut stream = conn.accept_uni().await.unwrap(); + + // Read until the reset surfaces, accumulating the delivered bytes. + let mut received = Vec::new(); + let code = loop { + match stream.read_chunk(usize::MAX).await { + Ok(Some(chunk)) => received.extend_from_slice(&chunk), + Ok(None) => panic!("a reliable reset must surface a reset, not a clean finish"), + Err(crate::ReadError::Reset(code)) => break code, + Err(e) => panic!("unexpected read error: {e:?}"), + } + }; + + // The sender committed to delivering at least the reliable prefix; the stack may deliver more + // (up to the final size), but everything delivered is an in-order prefix of the stream. + assert!(received.len() >= RELIABLE && received.len() <= MSG.len()); + assert_eq!(received, &MSG[..received.len()]); + assert_eq!(code, crate::VarInt::from_u32(CODE)); + + server.await.unwrap(); +} + #[test] fn export_keying_material() { let _guard = subscribe();