Reliable reset

src/aioquic/quic/configuration.py

@@ -112,6 +112,7 @@ class QuicConfiguration:
     original_version: Optional[int] = None
     private_key: Any = None
     quantum_readiness_test: bool = False
+    reliable_stream_reset: bool = False
     supported_versions: List[int] = field(
         default_factory=lambda: [
             QuicProtocolVersion.VERSION_1,

src/aioquic/quic/connection.py

@@ -56,7 +56,7 @@
 )
 from .packet_builder import QuicDeliveryState, QuicPacketBuilder, QuicPacketBuilderStop
 from .recovery import QuicPacketRecovery, QuicPacketSpace
-from .stream import FinalSizeError, QuicStream, StreamFinishedError
+from .stream import FinalSizeError, QuicStream, StreamFinishedError, StreamStateError
 
 logger = logging.getLogger("quic")
 
@@ -105,6 +105,7 @@
 PATH_RESPONSE_FRAME_CAPACITY = 1 + 8
 PING_FRAME_CAPACITY = 1
 RESET_STREAM_FRAME_CAPACITY = 1 + 3 * UINT_VAR_MAX_SIZE
+RESET_STREAM_AT_FRAME_CAPACITY = 1 + 4 * UINT_VAR_MAX_SIZE
 RETIRE_CONNECTION_ID_CAPACITY = 1 + UINT_VAR_MAX_SIZE
 STOP_SENDING_FRAME_CAPACITY = 1 + 2 * UINT_VAR_MAX_SIZE
 STREAMS_BLOCKED_CAPACITY = 1 + UINT_VAR_MAX_SIZE
@@ -450,6 +451,7 @@ def __init__(
             0x1C: (self._handle_connection_close_frame, EPOCHS("IH01")),
             0x1D: (self._handle_connection_close_frame, EPOCHS("01")),
             0x1E: (self._handle_handshake_done_frame, EPOCHS("1")),
+            0x24: (self._handle_reset_stream_frame, EPOCHS("01")),
             0x30: (self._handle_datagram_frame, EPOCHS("01")),
             0x31: (self._handle_datagram_frame, EPOCHS("01")),
         }
@@ -2088,6 +2090,10 @@ def _handle_reset_stream_frame(
         stream_id = buf.pull_uint_var()
         error_code = buf.pull_uint_var()
         final_size = buf.pull_uint_var()
+        if frame_type == QuicFrameType.RESET_STREAM_AT:
+            reliable_size = buf.pull_uint_var()
+        else:
+            reliable_size = None
 
         # log frame
         if self._quic_logger is not None:
@@ -2124,15 +2130,27 @@ def _handle_reset_stream_frame(
             final_size,
         )
         try:
-            event = stream.receiver.handle_reset(
-                error_code=error_code, final_size=final_size
-            )
+            if frame_type == QuicFrameType.RESET_STREAM:
+                event = stream.receiver.handle_reset(
+                    error_code=error_code, final_size=final_size
+                )
+            else:
+                assert frame_type == QuicFrameType.RESET_STREAM_AT
+                event = stream.receiver.handle_reset_at(
+                    error_code=error_code, final_size=final_size, reliable_size=reliable_size
+                )
         except FinalSizeError as exc:
             raise QuicConnectionError(
                 error_code=QuicErrorCode.FINAL_SIZE_ERROR,
                 frame_type=frame_type,
                 reason_phrase=str(exc),
             )
+        except StreamStateError as exc:
+            raise QuicConnectionError(
+                error_code=QuicErrorCode.STREAM_STATE_ERROR,
+                frame_type=frame_type,
+                reason_phrase=str(exc),
+            )
         if event is not None:
             self._events.append(event)
         self._local_max_data.used += newly_received
@@ -2268,8 +2286,19 @@ def _handle_stream_frame(
                 frame_type=frame_type,
                 reason_phrase=str(exc),
             )
+        except StreamStateError as exc:
+            raise QuicConnectionError(
+                error_code=QuicErrorCode.STREAM_STATE_ERROR,
+                frame_type=frame_type,
+                reason_phrase=str(exc),
+            )
         if event is not None:
             self._events.append(event)
+
+        pending_reset_event = stream.receiver.get_pending_reset_event()
+        if pending_reset_event is not None:
+            self._events.append(pending_reset_event)
+
         self._local_max_data.used += newly_received
 
     def _handle_stream_data_blocked_frame(
@@ -2887,6 +2916,7 @@ def _serialize_transport_parameters(self) -> bytes:
                 chosen_version=self._version,
                 available_versions=self._configuration.supported_versions,
             ),
+            reliable_stream_reset=self._configuration.reliable_stream_reset,
         )
         if not self._is_client:
             quic_transport_parameters.original_destination_connection_id = (
@@ -3124,6 +3154,11 @@ def _write_application(
                         # STOP_SENDING
                         self._write_stop_sending_frame(builder=builder, stream=stream)
 
+                    if stream.sender.reset_at_pending:
+                        # RESET_STREAM_AT
+                        self._write_reset_stream_at_frame(
+                            builder=builder, stream=stream)
+
                     if stream.sender.reset_pending:
                         # RESET_STREAM
                         self._write_reset_stream_frame(builder=builder, stream=stream)
@@ -3476,6 +3511,23 @@ def _write_reset_stream_frame(
                 )
             )
 
+    def _write_reset_stream_at_frame(
+        self,
+        builder: QuicPacketBuilder,
+        stream: QuicStream,
+    ) -> None:
+        buf = builder.start_frame(
+            frame_type=QuicFrameType.RESET_STREAM_AT,
+            capacity=RESET_STREAM_AT_FRAME_CAPACITY,
+            handler=stream.sender.on_reset_at_delivery,
+        )
+        frame = stream.sender.get_reset_at_frame()
+        buf.push_uint_var(frame.stream_id)
+        buf.push_uint_var(frame.error_code)
+        buf.push_uint_var(frame.final_size)
+        buf.push_uint_var(frame.reliable_size)
+        # TODO: log frame
+
     def _write_retire_connection_id_frame(
         self, builder: QuicPacketBuilder, sequence_number: int
     ) -> None:

src/aioquic/quic/packet.py

@@ -373,6 +373,7 @@ class QuicTransportParameters:
     version_information: Optional[QuicVersionInformation] = None
     max_datagram_frame_size: Optional[int] = None
     quantum_readiness: Optional[bytes] = None
+    reliable_stream_reset: Optional[bool] = None
 
 
 PARAMS = {
@@ -398,6 +399,8 @@ class QuicTransportParameters:
     # extensions
     0x0020: ("max_datagram_frame_size", int),
     0x0C37: ("quantum_readiness", bytes),
+    # https://datatracker.ietf.org/doc/html/draft-ietf-quic-reliable-stream-reset-06#section-8.1
+    0x17F7586D2CB571: ("reliable_stream_reset", bool),
 }
 
 
@@ -556,6 +559,7 @@ class QuicFrameType(IntEnum):
     HANDSHAKE_DONE = 0x1E
     DATAGRAM = 0x30
     DATAGRAM_WITH_LENGTH = 0x31
+    RESET_STREAM_AT = 0x24
 
 
 NON_ACK_ELICITING_FRAME_TYPES = frozenset(
@@ -593,6 +597,14 @@ class QuicResetStreamFrame:
     stream_id: int
 
 
+@dataclass
+class QuicResetStreamAtFrame:
+    error_code: int
+    final_size: int
+    stream_id: int
+    reliable_size: int
+
+
 @dataclass
 class QuicStopSendingFrame:
     error_code: int