feat(transport): implement RESET_STREAM_AT frame

neqo-transport/src/connection/params.rs

@@ -507,6 +507,8 @@ impl ConnectionParameters {
         }
         tps.local_mut()
             .set_integer(MaxDatagramFrameSize, self.datagram_size);
+        tps.local_mut()
+            .set_empty(TransportParameterId::ResetStreamAt);
         Ok(tps)
     }
 }

neqo-transport/src/frame.rs

@@ -47,6 +47,7 @@ pub enum FrameType {
     StreamsBlockedUniDi = 0x17,
     NewConnectionId = 0x18,
     RetireConnectionId = 0x19,
+    ResetStreamAt = 0x24,
     PathChallenge = 0x1a,
     PathResponse = 0x1b,
     ConnectionCloseTransport = 0x1c,
@@ -165,6 +166,12 @@ pub enum Frame<'a> {
         application_error_code: AppError,
         final_size: u64,
     },
+    ResetStreamAt {
+        stream_id: StreamId,
+        application_error_code: AppError,
+        final_size: u64,
+        reliable_size: u64,
+    },
     StopSending {
         stream_id: StreamId,
         application_error_code: AppError,
@@ -254,6 +261,7 @@ impl<'a> Frame<'a> {
             Self::Ping => FrameType::Ping,
             Self::Ack { .. } => FrameType::Ack,
             Self::ResetStream { .. } => FrameType::ResetStream,
+            Self::ResetStreamAt { .. } => FrameType::ResetStreamAt,
             Self::StopSending { .. } => FrameType::StopSending,
             Self::Crypto { .. } => FrameType::Crypto,
             Self::NewToken { .. } => FrameType::NewToken,
@@ -294,6 +302,7 @@ impl<'a> Frame<'a> {
         matches!(
             self,
             Self::ResetStream { .. }
+                | Self::ResetStreamAt { .. }
                 | Self::StopSending { .. }
                 | Self::Stream { .. }
                 | Self::MaxData { .. }
@@ -520,6 +529,18 @@ impl<'a> Frame<'a> {
                     _ => return Err(Error::NoMoreData),
                 },
             }),
+            FrameType::ResetStreamAt => Ok(Self::ResetStreamAt {
+                stream_id: StreamId::from(dv(dec)?),
+                application_error_code: dv(dec)?,
+                final_size: match dec.decode_varint() {
+                    Some(v) => v,
+                    _ => return Err(Error::NoMoreData),
+                },
+                reliable_size: match dec.decode_varint() {
+                    Some(v) => v,
+                    _ => return Err(Error::NoMoreData),
+                },
+            }),
             FrameType::Ack => decode_ack(dec, false),
             FrameType::AckEcn => decode_ack(dec, true),
             FrameType::StopSending => Ok(Self::StopSending {

neqo-transport/src/qlog.rs

@@ -431,6 +431,21 @@ impl From<Frame<'_>> for QuicFrame {
                 length: None,
                 payload_length: None,
             },
+            Frame::ResetStreamAt {
+                stream_id,
+                application_error_code,
+                final_size,
+                reliable_size: _,
+            } => {
+                // TODO: Add definition to qlog crate.
+                Self::ResetStream {
+                    stream_id: stream_id.as_u64(),
+                    error_code: application_error_code,
+                    final_size,
+                    length: None,
+                    payload_length: None,
+                }
+            }
             Frame::StopSending {
                 stream_id,
                 application_error_code,

neqo-transport/src/send_stream.rs

@@ -1006,6 +1006,7 @@ impl SendStream {
         tokens: &mut recovery::Tokens,
         stats: &mut FrameStats,
     ) -> bool {
+        // TODO: Always use ResetStreamAt.
         if let State::ResetSent {
             final_size,
             err,

neqo-transport/src/stats.rs

@@ -30,6 +30,7 @@ pub struct FrameStats {
     pub crypto: usize,
     pub stream: usize,
     pub reset_stream: usize,
+    pub reset_stream_at: usize,
     pub stop_sending: usize,
 
     pub ping: usize,
@@ -70,8 +71,8 @@ impl Debug for FrameStats {
         )?;
         writeln!(
             f,
-            "    stream {} reset {} stop {}",
-            self.stream, self.reset_stream, self.stop_sending,
+            "    stream {} reset {} reset_at {} stop {}",
+            self.stream, self.reset_stream, self.reset_stream_at, self.stop_sending,
         )?;
         writeln!(
             f,
@@ -103,6 +104,7 @@ impl FrameStats {
             + self.crypto
             + self.stream
             + self.reset_stream
+            + self.reset_stream_at
             + self.stop_sending
             + self.ping
             + self.padding

neqo-transport/src/streams.rs

@@ -121,7 +121,18 @@ impl Streams {
             } => {
                 stats.reset_stream += 1;
                 if let (_, Some(rs)) = self.obtain_stream(*stream_id)? {
-                    rs.reset(*application_error_code, *final_size)?;
+                    rs.reset(*application_error_code, *final_size, None)?;
+                }
+            }
+            Frame::ResetStreamAt {
+                stream_id,
+                application_error_code,
+                final_size,
+                reliable_size,
+            } => {
+                stats.reset_stream_at += 1;
+                if let (_, Some(rs)) = self.obtain_stream(*stream_id)? {
+                    rs.reset(*application_error_code, *final_size, Some(*reliable_size))?;
                 }
             }
             Frame::StopSending {

neqo-transport/src/tparams.rs

@@ -6,6 +6,11 @@
 
 // Transport parameters. See -transport section 7.3.
 
+// TODO: When using 0-RTT, both endpoints MUST remember the value of this
+// transport parameter. This allows use of this extension in 0-RTT packets. When
+// the server accepts 0-RTT data, the server MUST NOT disable this extension on
+// the resumed connection.
+
 use std::{
     cell::RefCell,
     fmt::{self, Display, Formatter},
@@ -54,6 +59,7 @@ pub enum TransportParameterId {
     GreaseQuicBit = 0x2ab2,
     MinAckDelay = 0xff02_de1a,
     MaxDatagramFrameSize = 0x0020,
+    ResetStreamAt = 0x17f7586d2cb571,
     #[cfg(test)]
     TestTransportParameter = 0xce16,
 }
@@ -324,6 +330,7 @@ impl TransportParameter {
                 _ => return Err(Error::TransportParameter),
             },
             TransportParameterId::VersionInformation => Self::decode_versions(&mut d)?,
+            TransportParameterId::ResetStreamAt => Self::Empty,
             #[cfg(test)]
             TransportParameterId::TestTransportParameter => {
                 Self::Bytes(d.decode_remainder().to_vec())