Fix flow control leak on RESET_STREAM and process resets eagerly

We weren't issuing additional flow control credit to account for the
bytes that the peer thinks it's sent but which we haven't read yet. As
part of this change, we've gone from keeping data on reset streams
around until it's read and issuing credit as that proceeds to
discarding data on reset streams immediately and issuing credit for
all outstanding data on the stream, whether or not we've received and
buffered it.

quinn-proto/src/connection.rs

@@ -169,13 +169,13 @@ impl Connection {
         for i in 0..config.stream_window_uni {
             streams.insert(
                 StreamId::new(!side, Directionality::Uni, u64::from(i)),
-                stream::Recv::new(u64::from(config.stream_receive_window)).into(),
+                stream::Recv::new().into(),
             );
         }
         for i in 0..config.stream_window_bidi {
             streams.insert(
                 StreamId::new(!side, Directionality::Bi, i as u64),
-                Stream::new_bi(config.stream_receive_window as u64),
+                Stream::new_bi(),
             );
         }
         let mut loc_cids = HashMap::new();
@@ -1457,6 +1457,10 @@ impl Connection {
                         _ => {}
                     }
                     let rs = self.streams.get_recv_mut(frame.id).unwrap();
+                    if rs.is_finished() {
+                        trace!(self.log, "dropping frame for finished stream");
+                        continue;
+                    }
 
                     let end = frame.offset + frame.data.len() as u64;
                     if let Some(final_offset) = rs.final_offset() {
@@ -1467,10 +1471,11 @@ impl Connection {
                     }
                     let prev_end = rs.limit();
                     let new_bytes = end.saturating_sub(prev_end);
-                    if end > rs.max_data || data_recvd + new_bytes > max_data {
+                    let stream_max_data = rs.bytes_read + self.config.stream_receive_window;
+                    if end > stream_max_data || data_recvd + new_bytes > max_data {
                         debug!(self.log, "flow control error";
                                    "stream" => frame.id.0, "recvd" => data_recvd, "new bytes" => new_bytes,
-                                   "max data" => max_data, "end" => end, "stream max data" => rs.max_data);
+                                   "max data" => max_data, "end" => end, "stream max data" => stream_max_data);
                         return Err(TransportError::FLOW_CONTROL_ERROR);
                     }
                     if frame.fin {
@@ -1594,7 +1599,7 @@ impl Connection {
                     error_code,
                     final_offset,
                 }) => {
-                    let (offset, fresh) = match self.streams.get_recv_stream(self.side, id) {
+                    let rs = match self.streams.get_recv_stream(self.side, id) {
                         Err(e) => {
                             debug!(self.log, "received illegal RST_STREAM");
                             return Err(e);
@@ -1603,26 +1608,32 @@ impl Connection {
                             trace!(self.log, "received RST_STREAM on closed stream");
                             continue;
                         }
-                        Ok(Some(stream)) => {
-                            let rs = stream.recv_mut().unwrap();
-                            let limit = rs.limit();
-                            if let Some(offset) = rs.final_offset() {
-                                if offset != final_offset {
-                                    return Err(TransportError::FINAL_OFFSET_ERROR);
-                                }
-                            } else if limit > final_offset {
-                                return Err(TransportError::FINAL_OFFSET_ERROR);
-                            }
-                            if !rs.is_closed() {
-                                rs.state = stream::RecvState::ResetRecvd {
-                                    size: final_offset,
-                                    error_code,
-                                };
-                            }
-                            (limit, mem::replace(&mut rs.fresh, false))
-                        }
+                        Ok(Some(stream)) => stream.recv_mut().unwrap(),
                     };
-                    self.data_recvd += final_offset - offset;
+                    let limit = rs.limit();
+                    let fresh = mem::replace(&mut rs.fresh, false);
+
+                    // Validate final_offset
+                    if let Some(offset) = rs.final_offset() {
+                        if offset != final_offset {
+                            return Err(TransportError::FINAL_OFFSET_ERROR);
+                        }
+                    } else if limit > final_offset {
+                        return Err(TransportError::FINAL_OFFSET_ERROR);
+                    }
+
+                    // State transition
+                    rs.reset(error_code, final_offset);
+
+                    // Update flow control
+                    if rs.bytes_read != final_offset {
+                        self.data_recvd += final_offset - limit;
+                        // bytes_read is always <= limit, so this won't underflow.
+                        self.local_max_data += final_offset - rs.bytes_read;
+                        self.space_mut(SpaceId::Data).pending.max_data = true;
+                    }
+
+                    // Notify application
                     if fresh {
                         self.stream_opened = true;
                         self.streams.incoming.push_back(id);
@@ -1942,10 +1953,16 @@ impl Connection {
                 continue;
             }
             sent.max_stream_data.insert(id);
-            trace!(self.log, "MAX_STREAM_DATA"; "stream" => id.0, "value" => rs.max_data);
+            let max = rs.bytes_read + self.config.stream_receive_window;
+            trace!(
+                self.log,
+                "MAX_STREAM_DATA: {stream} = {max}",
+                stream = id,
+                max = max
+            );
             buf.write(frame::Type::MAX_STREAM_DATA);
             buf.write(id);
-            buf.write_var(rs.max_data);
+            buf.write_var(max);
         }
 
         // MAX_STREAMS_UNI
@@ -2314,7 +2331,7 @@ impl Connection {
                 self.streams.next_bi += 1;
                 (
                     StreamId::new(self.side, direction, self.streams.next_bi - 1),
-                    Stream::new_bi(self.config.stream_receive_window as u64),
+                    Stream::new_bi(),
                 )
             }
             _ => {
@@ -2364,7 +2381,7 @@ impl Connection {
                         Directionality::Bi,
                         self.streams.max_remote_bi - 1,
                     ),
-                    Stream::new_bi(self.config.stream_receive_window as u64),
+                    Stream::new_bi(),
                 )
             }
             Directionality::Uni => {
@@ -2376,7 +2393,7 @@ impl Connection {
                         Directionality::Uni,
                         self.streams.max_remote_uni - 1,
                     ),
-                    stream::Recv::new(self.config.stream_receive_window).into(),
+                    stream::Recv::new().into(),
                 )
             }
         };
@@ -2424,6 +2441,7 @@ impl Connection {
         let space = &mut self.spaces[SpaceId::Data as usize];
         space.pending.max_data = true;
         if rs.receiving_unknown_size() {
+            // Only bother issuing stream credit if the peer wants to send more
             space.pending.max_stream_data.insert(id);
         }
         Ok((buf, len))
@@ -2441,6 +2459,7 @@ impl Connection {
         let space = &mut self.spaces[SpaceId::Data as usize];
         space.pending.max_data = true;
         if rs.receiving_unknown_size() {
+            // Only bother issuing stream credit if the peer wants to send more
             space.pending.max_stream_data.insert(id);
         }
         Ok(len)

quinn-proto/src/stream.rs

@@ -12,8 +12,8 @@ pub enum Stream {
 }
 
 impl Stream {
-    pub fn new_bi(window: u64) -> Self {
-        Stream::Both(Send::new(), Recv::new(window))
+    pub fn new_bi() -> Self {
+        Stream::Both(Send::new(), Recv::new())
     }
 
     pub fn send(&self) -> Option<&Send> {
@@ -132,26 +132,27 @@ pub struct Recv {
     pub state: RecvState,
     pub recvd: RangeSet,
     pub buffered: VecDeque<(Bytes, u64)>,
-    /// Upper limit dictated by the peer
-    pub max_data: u64,
     /// Whether any unordered reads have been performed, making this stream unusable for ordered
     /// reads
     pub unordered: bool,
     pub assembler: Assembler,
     /// Whether the application has been notified of this stream yet
     pub fresh: bool,
+    /// Number of bytes read by the application. Equal to assembler.offset when `unordered` is
+    /// false.
+    pub bytes_read: u64,
 }
 
 impl Recv {
-    pub fn new(max_data: u64) -> Self {
+    pub fn new() -> Self {
         Self {
             state: RecvState::Recv { size: None },
             recvd: RangeSet::new(),
             buffered: VecDeque::new(),
-            max_data,
             unordered: false,
             assembler: Assembler::new(),
             fresh: true,
+            bytes_read: 0,
         }
     }
 
@@ -167,10 +168,7 @@ impl Recv {
 
         if !self.assembler.blocked() {
             let n = self.assembler.read(buf);
-            // Only bother issuing stream credit if the peer wants to send more
-            if self.receiving_unknown_size() {
-                self.max_data += n as u64;
-            }
+            self.bytes_read += n as u64;
             Ok(n)
         } else {
             match self.state {
@@ -194,10 +192,7 @@ impl Recv {
 
         // Return data we already have buffered, regardless of state
         if let Some(x) = self.buffered.pop_front() {
-            // Only bother issuing stream credit if the peer wants to send more
-            if self.receiving_unknown_size() {
-                self.max_data += x.0.len() as u64;
-            }
+            self.bytes_read += x.0.len() as u64;
             Ok(x)
         } else {
             match self.state {
@@ -256,6 +251,22 @@ impl Recv {
             _ => None,
         }
     }
+
+    pub fn reset(&mut self, error_code: u16, final_offset: u64) {
+        if self.is_closed() {
+            return;
+        }
+        self.state = RecvState::ResetRecvd {
+            size: final_offset,
+            error_code,
+        };
+        // Nuke buffers so that future reads fail immediately, which ensures future reads don't
+        // issue flow control credit redundant to that already issued. We could instead special-case
+        // reset streams during read, but it's unclear if there's any benefit to retaining data for
+        // reset streams.
+        self.buffered.clear();
+        self.assembler.clear();
+    }
 }
 
 #[derive(Debug, Fail, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
@@ -399,6 +410,12 @@ impl Assembler {
     pub fn offset(&self) -> u64 {
         self.offset
     }
+
+    /// Discard all buffered data
+    pub fn clear(&mut self) {
+        self.written = VecDeque::new();
+        self.data = VecDeque::new();
+    }
 }
 
 #[cfg(test)]

quinn-proto/src/tests.rs

@@ -546,7 +546,6 @@ fn reset_stream() {
     assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn);
     assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s);
     assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s);
-    assert_matches!(pair.server.read_unordered(server_conn, s), Ok((ref data, 0)) if data == MSG);
     assert_matches!(
         pair.server.read_unordered(server_conn, s),
         Err(ReadError::Reset { error_code: ERROR })
@@ -572,7 +571,6 @@ fn stop_stream() {
     assert_matches!(pair.server.poll(), Some((conn, Event::StreamOpened)) if conn == server_conn);
     assert_matches!(pair.server.accept_stream(server_conn), Some(stream) if stream == s);
     assert_matches!(pair.server.poll(), Some((conn, Event::StreamReadable { stream })) if conn == server_conn && stream == s);
-    assert_matches!(pair.server.read_unordered(server_conn, s), Ok((ref data, 0)) if data == MSG);
     assert_matches!(
         pair.server.read_unordered(server_conn, s),
         Err(ReadError::Reset { error_code: ERROR })