wip: thanks to claude

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<Chunk> {
+        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<Chunk> {
         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;

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)),
         }
     }
 }

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<u8> {
         buf.data.to_vec()
     }
@@ -636,6 +747,8 @@ mod proptests {
         Retransmit(Range<u64>),
         // 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);

noq-proto/src/connection/spaces.rs

@@ -558,8 +558,15 @@ pub struct Retransmits {
     pub(super) remove_address: BTreeSet<RemoveAddress>,
     /// 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 {

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)]

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<u64> {
         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<ResetAtOutcome, TransportError> {
+        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<VarInt> {
         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<u64> {
+        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<IllegalOrderedRead> 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<u64> },
-    ResetRecvd { size: u64, error_code: VarInt },
+    Recv {
+        size: Option<u64>,
+    },
+    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);
+    }
 }

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<VarInt>,
+    /// 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<ResetAt>,
+}
+
+/// 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<bool, ResetStreamAtError> {
+        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<Send> {
+        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();

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<ShouldTransmit, TransportError> {
+        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"
+        );
+    }
 }

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<Value = VarInt> {
+    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<W: BufMut>(&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

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);
+}

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)
+        );
+    }
 }

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};

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(())
     }
 

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();