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

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

@@ -82,7 +82,7 @@ where
     ///
     /// The peer will receive a request error with `REQUEST_CANCELLED` code.
     pub fn cancel(&mut self) {
-        self.send.reset(ErrorCode::REQUEST_CANCELLED.into());
+        let _ = self.send.reset(ErrorCode::REQUEST_CANCELLED.into());
         self.state = SendDataState::Finished;
     }
 }

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

@@ -1307,24 +1307,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
@@ -2109,7 +2115,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

@@ -351,15 +351,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
@@ -374,6 +378,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!(
@@ -869,7 +869,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

@@ -136,17 +136,17 @@ where
 
     /// Close the send stream immediately.
     ///
-    /// 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 `poll_finish`
-    /// was called previously and all data has already been transmitted at least once, the peer
-    /// may still receive all written data.
-    pub fn reset(&mut self, error_code: VarInt) {
+    /// 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) -> 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)]