Don't panic from unknown StreamIds

quinn-proto/src/connection.rs

@@ -23,7 +23,7 @@ use crate::shared::{
     EndpointEvent, EndpointEventInner, IssuedCid, ServerConfig, TransportConfig,
 };
 use crate::spaces::{CryptoSpace, PacketSpace, Retransmits, SentPacket};
-use crate::streams::{self, FinishError, ReadError, Streams, WriteError};
+use crate::streams::{self, FinishError, ReadError, Streams, UnknownStream, WriteError};
 use crate::timer::{Timer, TimerKind, TimerTable};
 use crate::transport_parameters::{self, TransportParameters};
 use crate::{
@@ -818,8 +818,11 @@ where
         );
     }
 
-    fn queue_stream_data(&mut self, stream: StreamId, data: Bytes) {
-        let ss = self.streams.send_mut(stream).unwrap();
+    fn queue_stream_data(&mut self, stream: StreamId, data: Bytes) -> Result<(), WriteError> {
+        let ss = self
+            .streams
+            .send_mut(stream)
+            .ok_or(WriteError::UnknownStream)?;
         assert_eq!(ss.state, streams::SendState::Ready);
         let offset = ss.offset;
         ss.offset += data.len() as u64;
@@ -835,6 +838,7 @@ where
                 data,
                 id: stream,
             });
+        Ok(())
     }
 
     /// Abandon transmitting data on a stream
@@ -2656,12 +2660,15 @@ where
     }
 
     /// Signal to the peer that it should stop sending on the given recv stream
-    pub fn stop_sending(&mut self, id: StreamId, error_code: VarInt) {
+    pub fn stop_sending(&mut self, id: StreamId, error_code: VarInt) -> Result<(), UnknownStream> {
         assert!(
             id.directionality() == Directionality::Bi || id.initiator() != self.side,
             "only streams supporting incoming data may be stopped"
         );
-        let stream = self.streams.recv_mut(id).unwrap();
+        let stream = self
+            .streams
+            .recv_mut(id)
+            .ok_or(UnknownStream { _private: () })?;
         // Only bother if there's data we haven't received yet
         if !stream.is_finished() {
             let space = &mut self.spaces[SpaceId::Data as usize];
@@ -2670,6 +2677,7 @@ where
                 .stop_sending
                 .push(frame::StopSending { id, error_code });
         }
+        Ok(())
     }
 
     fn congestion_blocked(&self) -> bool {
@@ -2846,7 +2854,7 @@ where
             self.config.send_window - self.unacked_data,
         );
         let n = conn_budget.min(stream_budget).min(data.len() as u64) as usize;
-        self.queue_stream_data(stream, (&data[0..n]).into());
+        self.queue_stream_data(stream, (&data[0..n]).into())?;
         trace!(
             self.log,
             "wrote {len} bytes to {stream}",

quinn-proto/src/lib.rs

@@ -58,7 +58,7 @@ pub use crate::shared::{
 };
 
 mod streams;
-pub use crate::streams::{FinishError, ReadError, WriteError};
+pub use crate::streams::{FinishError, ReadError, UnknownStream, WriteError};
 
 mod transport_error;
 pub use crate::transport_error::{Code as TransportErrorCode, Error as TransportError};

quinn-proto/src/streams.rs

@@ -513,3 +513,9 @@ pub enum FinishError {
     #[error(display = "unknown stream")]
     UnknownStream,
 }
+
+/// Unknown stream ID
+#[derive(Debug)]
+pub struct UnknownStream {
+    pub(crate) _private: (),
+}

quinn-proto/src/tests/mod.rs

@@ -272,7 +272,9 @@ fn stop_stream() {
 
     info!(pair.log, "stopping stream");
     const ERROR: VarInt = VarInt(42);
-    pair.server_conn_mut(server_ch).stop_sending(s, ERROR);
+    pair.server_conn_mut(server_ch)
+        .stop_sending(s, ERROR)
+        .unwrap();
     pair.drive();
 
     assert_matches!(
@@ -904,7 +906,8 @@ fn stop_opens_bidi() {
         .connections
         .get_mut(&server_conn)
         .unwrap()
-        .stop_sending(s, ERROR);
+        .stop_sending(s, ERROR)
+        .unwrap();
     pair.drive();
 
     assert_matches!(
@@ -1090,7 +1093,9 @@ fn stop_before_finish() {
 
     info!(pair.log, "stopping stream");
     const ERROR: VarInt = VarInt(42);
-    pair.server_conn_mut(server_ch).stop_sending(s, ERROR);
+    pair.server_conn_mut(server_ch)
+        .stop_sending(s, ERROR)
+        .unwrap();
     pair.drive();
 
     assert_matches!(
@@ -1115,7 +1120,9 @@ fn stop_during_finish() {
     assert_matches!(pair.server_conn_mut(server_ch).accept(), Some(stream) if stream == s);
     info!(pair.log, "stopping and finishing stream");
     const ERROR: VarInt = VarInt(42);
-    pair.server_conn_mut(server_ch).stop_sending(s, ERROR);
+    pair.server_conn_mut(server_ch)
+        .stop_sending(s, ERROR)
+        .unwrap();
     pair.drive_server();
     pair.client_conn_mut(client_ch).finish(s).unwrap();
     pair.drive_client();

quinn/src/streams.rs

@@ -6,7 +6,7 @@ use err_derive::Error;
 use futures::sync::oneshot;
 use futures::{task, try_ready};
 use futures::{Async, Future, Poll};
-use proto::{ConnectionError, StreamId};
+use proto::{self, ConnectionError, StreamId};
 use tokio_io::{AsyncRead, AsyncWrite};
 
 use crate::connection::ConnectionRef;
@@ -333,14 +333,15 @@ impl RecvStream {
     ///
     /// Has no effect if the incoming stream already finished, even if the local application hasn't
     /// yet read all buffered data.
-    pub fn stop(&mut self, error_code: VarInt) {
+    pub fn stop(&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.stop_sending(self.stream, error_code);
+        conn.inner.stop_sending(self.stream, error_code)?;
         conn.notify();
         self.all_data_read = true;
+        Ok(())
     }
 }
 
@@ -422,7 +423,8 @@ impl Drop for RecvStream {
             return;
         }
         if !self.all_data_read {
-            conn.inner.stop_sending(self.stream, 0u32.into());
+            // Ignore UnknownStream errors
+            let _ = conn.inner.stop_sending(self.stream, 0u32.into());
             conn.notify();
         }
     }
@@ -499,3 +501,12 @@ impl From<WriteError> for io::Error {
         io::Error::new(kind, x)
     }
 }
+
+#[derive(Debug)]
+pub struct UnknownStream {}
+
+impl From<proto::UnknownStream> for UnknownStream {
+    fn from(_: proto::UnknownStream) -> Self {
+        UnknownStream {}
+    }
+}