Return UnknownStream from Connection::reset for closed/reset streams

Improves consistency with other stream interfaces. H3 logic was
updated to ignore errors to preserve existing behavior, but future
work might want to insert some unwraps there if static correctness
guarantees are available.

Author: Ralith

quinn-h3/src/connection.rs

@@ -233,7 +233,7 @@ impl ConnectionInner {
                     }
                     Side::Server => {
                         if self.inner.is_closing() {
-                            send.reset(ErrorCode::REQUEST_REJECTED.into());
+                            let _ = send.reset(ErrorCode::REQUEST_REJECTED.into());
                             let _ = recv.stop(ErrorCode::REQUEST_REJECTED.into());
                         } else {
                             self.inner.request_initiated(send.id());

quinn-h3/src/data.rs

@@ -92,7 +92,7 @@ where
             }
             _ => {
                 if let Some(ref mut send) = self.send.take() {
-                    send.reset(ErrorCode::REQUEST_CANCELLED.into());
+                    let _ = send.reset(ErrorCode::REQUEST_CANCELLED.into());
                 }
             }
         }

quinn-h3/src/frame.rs

@@ -143,7 +143,7 @@ where
 
     pub fn reset(&mut self, err_code: ErrorCode) {
         if let Some(ref mut s) = self.send {
-            s.reset(err_code.into());
+            let _ = s.reset(err_code.into());
         }
     }
 }

quinn-h3/src/server.rs

@@ -573,7 +573,7 @@ impl RecvRequest {
             recv.reset(ErrorCode::REQUEST_REJECTED);
         }
         if let Some(mut send) = self.send.take() {
-            send.reset(ErrorCode::REQUEST_REJECTED.into());
+            let _ = send.reset(ErrorCode::REQUEST_REJECTED.into());
         }
     }
 
@@ -604,7 +604,7 @@ impl RecvRequest {
                 r.reset(ErrorCode::REQUEST_REJECTED);
             }
             if let Some(s) = &mut self.send {
-                s.reset(ErrorCode::REQUEST_REJECTED.into());
+                let _ = s.reset(ErrorCode::REQUEST_REJECTED.into());
             }
             return Err(Error::peer(format!(
                 "Tried an non indempotent method in 0-RTT: {}",
@@ -718,7 +718,8 @@ impl Sender {
     /// Cancelling a request means that some request data have been processed by the application, which
     /// decided to abandon the response.
     pub fn cancel(&mut self) {
-        self.send
+        let _ = self
+            .send
             .as_mut()
             .unwrap()
             .reset(ErrorCode::REQUEST_CANCELLED.into());

quinn-proto/src/connection/mod.rs

@@ -1304,24 +1304,30 @@ where
     ///
     /// # Panics
     /// - when applied to a receive stream or an unopened send stream
-    pub fn reset(&mut self, stream_id: StreamId, error_code: VarInt) {
-        self.reset_inner(stream_id, error_code, false);
+    pub fn reset(&mut self, stream_id: StreamId, error_code: VarInt) -> Result<(), UnknownStream> {
+        self.reset_inner(stream_id, error_code, false)
     }
 
     /// `stopped` should be set iff this is an internal implicit reset due to `STOP_SENDING`
-    fn reset_inner(&mut self, stream_id: StreamId, error_code: VarInt, stopped: bool) {
+    fn reset_inner(
+        &mut self,
+        stream_id: StreamId,
+        error_code: VarInt,
+        stopped: bool,
+    ) -> Result<(), UnknownStream> {
         assert!(
             stream_id.dir() == Dir::Bi || stream_id.initiator() == self.side,
             "only streams supporting outgoing data may be reset"
         );
 
         let stop_reason = if stopped { Some(error_code) } else { None };
-        self.streams.reset(stream_id, stop_reason);
+        self.streams.reset(stream_id, stop_reason)?;
 
         self.spaces[SpaceId::Data as usize]
             .pending
             .reset_stream
             .push((stream_id, error_code));
+        Ok(())
     }
 
     /// Handle the already-decrypted first packet from the client
@@ -2106,7 +2112,8 @@ where
                             "STOP_SENDING on unopened stream",
                         ));
                     }
-                    self.reset_inner(id, error_code, true);
+                    // Ignore errors from the stream already being gone
+                    let _ = self.reset_inner(id, error_code, true);
                 }
                 Frame::RetireConnectionId { sequence } => {
                     if self.endpoint_config.local_cid_len == 0 {

quinn-proto/src/connection/streams.rs

@@ -344,15 +344,19 @@ impl Streams {
     ///
     /// Does not cause the actual RESET_STREAM frame to be sent, just updates internal
     /// state.
-    pub fn reset(&mut self, id: StreamId, stop_reason: Option<VarInt>) {
+    pub fn reset(
+        &mut self,
+        id: StreamId,
+        stop_reason: Option<VarInt>,
+    ) -> Result<(), UnknownStream> {
         let stream = match self.send.get_mut(&id) {
             Some(ss) => ss,
-            None => return,
+            None => return Err(UnknownStream { _private: () }),
         };
 
         if matches!(stream.state, SendState::ResetSent { .. } | SendState::ResetRecvd { .. }) {
-            // Ignore redundant reset calls
-            return;
+            // Redundant reset call
+            return Err(UnknownStream { _private: () });
         }
 
         // Restore the portion of the send window consumed by the data that we aren't about to
@@ -367,6 +371,8 @@ impl Streams {
         if stop_reason.is_some() && !stream.is_closed() {
             self.on_stream_frame(false, id);
         }
+
+        Ok(())
     }
 
     pub fn reset_acked(&mut self, id: StreamId) {

quinn-proto/src/tests/mod.rs

@@ -231,7 +231,7 @@ fn reset_stream() {
 
     info!("resetting stream");
     const ERROR: VarInt = VarInt(42);
-    pair.client_conn_mut(client_ch).reset(s, ERROR);
+    pair.client_conn_mut(client_ch).reset(s, ERROR).unwrap();
     pair.drive();
 
     assert_matches!(
@@ -811,7 +811,9 @@ fn test_flow_control(config: TransportConfig, window_size: usize) {
         Err(WriteError::Blocked)
     );
     pair.drive();
-    pair.client_conn_mut(client_conn).reset(s, VarInt(42));
+    pair.client_conn_mut(client_conn)
+        .reset(s, VarInt(42))
+        .unwrap();
     pair.drive();
     assert_eq!(
         pair.server_conn_mut(server_conn).read(s, &mut buf),

quinn/src/streams.rs

@@ -139,13 +139,14 @@ where
     /// No new data can be written after calling this method. Locally buffered data is dropped, and
     /// previously transmitted data will no longer be retransmitted if lost. If an attempt has
     /// already been made to finish the stream, the peer may still receive all written data.
-    pub fn reset(&mut self, error_code: VarInt) {
+    pub fn reset(&mut self, error_code: VarInt) -> Result<(), UnknownStream> {
         let mut conn = self.conn.lock().unwrap();
         if self.is_0rtt && conn.check_0rtt().is_err() {
-            return;
+            return Ok(());
         }
-        conn.inner.reset(self.stream, error_code);
+        conn.inner.reset(self.stream, error_code)?;
         conn.wake();
+        Ok(())
     }
 
     #[doc(hidden)]