Send RST_STREAM of STREAM_CLOSED instead of GOAWAY if stream may have been forgotten

ci/h2spec.sh

@@ -3,7 +3,7 @@ LOGFILE="h2server.log"
 
 if ! [ -e "h2spec" ] ; then
     # if we don't already have a h2spec executable, wget it from github
-    wget https://github.com/summerwind/h2spec/releases/download/v2.1.0/h2spec_linux_amd64.tar.gz
+    wget https://github.com/summerwind/h2spec/releases/download/v2.1.1/h2spec_linux_amd64.tar.gz
     tar xf h2spec_linux_amd64.tar.gz
 fi
 

src/proto/streams/recv.rs

@@ -516,28 +516,19 @@ impl Recv {
             stream.recv_flow.window_size()
         );
 
-        // Ensure that there is enough capacity on the connection before acting
-        // on the stream.
-        self.consume_connection_window(sz)?;
 
         if is_ignoring_frame {
             trace!(
                 "recv_data; frame ignored on locally reset {:?} for some time",
                 stream.id,
             );
-            // we just checked for enough connection window capacity, and
+            return self.ignore_data(sz);
-            // consumed it. Since we are ignoring this frame "for some time",
-            // we aren't returning the frame to the user. That means they
-            // have no way to release the capacity back to the connection. So
-            // we have to release it automatically.
-            //
-            // This call doesn't send a WINDOW_UPDATE immediately, just marks
-            // the capacity as available to be reclaimed. When the available
-            // capacity meets a threshold, a WINDOW_UPDATE is then sent.
-            self.release_connection_capacity(sz, &mut None);
-            return Ok(());
         }
 
+        // Ensure that there is enough capacity on the connection before acting
+        // on the stream.
+        self.consume_connection_window(sz)?;
+
         if stream.recv_flow.window_size() < sz {
             // http://httpwg.org/specs/rfc7540.html#WINDOW_UPDATE
             // > A receiver MAY respond with a stream error (Section 5.4.2) or
@@ -599,6 +590,22 @@ impl Recv {
         Ok(())
     }
 
+    pub fn ignore_data(&mut self, sz: WindowSize) -> Result<(), RecvError> {
+        // Ensure that there is enough capacity on the connection...
+        self.consume_connection_window(sz)?;
+
+        // Since we are ignoring this frame,
+        // we aren't returning the frame to the user. That means they
+        // have no way to release the capacity back to the connection. So
+        // we have to release it automatically.
+        //
+        // This call doesn't send a WINDOW_UPDATE immediately, just marks
+        // the capacity as available to be reclaimed. When the available
+        // capacity meets a threshold, a WINDOW_UPDATE is then sent.
+        self.release_connection_capacity(sz, &mut None);
+        return Ok(());
+    }
+
     pub fn consume_connection_window(&mut self, sz: WindowSize) -> Result<(), RecvError> {
         if self.flow.window_size() < sz {
             debug!(
@@ -753,7 +760,7 @@ impl Recv {
         self.max_stream_id
     }
 
-    fn next_stream_id(&self) -> Result<StreamId, RecvError> {
+    pub fn next_stream_id(&self) -> Result<StreamId, RecvError> {
         if let Ok(id) = self.next_stream_id {
             Ok(id)
         } else {
@@ -761,6 +768,19 @@ impl Recv {
         }
     }
 
+    pub fn may_have_created_stream(&self, id: StreamId) -> bool {
+        if let Ok(next_id) = self.next_stream_id {
+            // Peer::is_local_init should have been called beforehand
+            debug_assert_eq!(
+                id.is_server_initiated(),
+                next_id.is_server_initiated(),
+            );
+            id < next_id
+        } else {
+            true
+        }
+    }
+
     /// Returns true if the remote peer can reserve a stream with the given ID.
     pub fn ensure_can_reserve(&self)
         -> Result<(), RecvError>

src/proto/streams/send.rs

@@ -109,12 +109,6 @@ impl Send {
     }
 
     /// Send an explicit RST_STREAM frame
-    ///
-    /// # Arguments
-    /// + `reason`: the error code for the RST_STREAM frame
-    /// + `clear_queue`: if true, all pending outbound frames will be cleared,
-    ///    if false, the RST_STREAM frame will be appended to the end of the
-    ///    send queue.
     pub fn send_reset<B>(
         &mut self,
         reason: Reason,
@@ -452,4 +446,17 @@ impl Send {
     pub fn ensure_next_stream_id(&self) -> Result<StreamId, UserError> {
         self.next_stream_id.map_err(|_| UserError::OverflowedStreamId)
     }
+
+    pub fn may_have_created_stream(&self, id: StreamId) -> bool {
+        if let Ok(next_id) = self.next_stream_id {
+            // Peer::is_local_init should have been called beforehand
+            debug_assert_eq!(
+                id.is_server_initiated(),
+                next_id.is_server_initiated(),
+            );
+            id < next_id
+        } else {
+            true
+        }
+    }
 }

src/proto/streams/streams.rs

@@ -140,17 +140,39 @@ where
 
         let key = match me.store.find_entry(id) {
             Entry::Occupied(e) => e.key(),
-            Entry::Vacant(e) => match me.actions.recv.open(id, Open::Headers, &mut me.counts)? {
+            Entry::Vacant(e) => {
-                Some(stream_id) => {
+                // Client: it's possible to send a request, and then send
-                    let stream = Stream::new(
+                // a RST_STREAM while the response HEADERS were in transit.
-                        stream_id,
+                //
-                        me.actions.send.init_window_sz(),
+                // Server: we can't reset a stream before having received
-                        me.actions.recv.init_window_sz(),
+                // the request headers, so don't allow.
-                    );
+                if !P::is_server() {
+                    // This may be response headers for a stream we've already
+                    // forgotten about...
+                    if me.actions.may_have_forgotten_stream::<P>(id) {
+                        debug!(
+                            "recv_headers for old stream={:?}, sending STREAM_CLOSED",
+                            id,
+                        );
+                        return Err(RecvError::Stream {
+                            id,
+                            reason: Reason::STREAM_CLOSED,
+                        });
+                    }
+                }
 
-                    e.insert(stream)
+                match me.actions.recv.open(id, Open::Headers, &mut me.counts)? {
-                },
+                    Some(stream_id) => {
-                None => return Ok(()),
+                        let stream = Stream::new(
+                            stream_id,
+                            me.actions.send.init_window_sz(),
+                            me.actions.recv.init_window_sz(),
+                        );
+
+                        e.insert(stream)
+                    },
+                    None => return Ok(()),
+                }
             },
         };
 
@@ -236,6 +258,25 @@ where
                     return Ok(());
                 }
 
+                if me.actions.may_have_forgotten_stream::<P>(id) {
+                    debug!(
+                        "recv_data for old stream={:?}, sending STREAM_CLOSED",
+                        id,
+                    );
+
+                    let sz = frame.payload().len();
+                    // This should have been enforced at the codec::FramedRead layer, so
+                    // this is just a sanity check.
+                    assert!(sz <= super::MAX_WINDOW_SIZE as usize);
+                    let sz = sz as WindowSize;
+
+                    me.actions.recv.ignore_data(sz)?;
+                    return Err(RecvError::Stream {
+                        id,
+                        reason: Reason::STREAM_CLOSED,
+                    });
+                }
+
                 proto_err!(conn: "recv_data: stream not found; id={:?}", id);
                 return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
             },
@@ -674,13 +715,10 @@ where
 
         let key = match me.store.find_entry(id) {
             Entry::Occupied(e) => e.key(),
-            Entry::Vacant(e) => match me.actions.recv.open(id, Open::Headers, &mut me.counts) {
+            Entry::Vacant(e) => {
-                Ok(Some(stream_id)) => {
+                let stream = Stream::new(id, 0, 0);
-                    let stream = Stream::new(stream_id, 0, 0);
 
-                    e.insert(stream)
+                e.insert(stream)
-                },
-                _ => return,
             },
         };
 
@@ -1250,6 +1288,26 @@ impl Actions {
         }
     }
 
+    /// Check if we possibly could have processed and since forgotten this stream.
+    ///
+    /// If we send a RST_STREAM for a stream, we will eventually "forget" about
+    /// the stream to free up memory. It's possible that the remote peer had
+    /// frames in-flight, and by the time we receive them, our own state is
+    /// gone. We *could* tear everything down by sending a GOAWAY, but it
+    /// is more likely to be latency/memory constraints that caused this,
+    /// and not a bad actor. So be less catastrophic, the spec allows
+    /// us to send another RST_STREAM of STREAM_CLOSED.
+    fn may_have_forgotten_stream<P: Peer>(&self, id: StreamId) -> bool {
+        if id.is_zero() {
+            return false;
+        }
+        if P::is_local_init(id) {
+            self.send.may_have_created_stream(id)
+        } else {
+            self.recv.may_have_created_stream(id)
+        }
+    }
+
     fn clear_queues(&mut self,
                     clear_pending_accept: bool,
                     store: &mut Store,

tests/h2-support/src/frames.rs

@@ -316,6 +316,11 @@ impl Mock<frame::Reset> {
         Mock(frame::Reset::new(id, frame::Reason::CANCEL))
     }
 
+    pub fn stream_closed(self) -> Self {
+        let id = self.0.stream_id();
+        Mock(frame::Reset::new(id, frame::Reason::STREAM_CLOSED))
+    }
+
     pub fn internal_error(self) -> Self {
         let id = self.0.stream_id();
         Mock(frame::Reset::new(id, frame::Reason::INTERNAL_ERROR))

tests/h2-tests/tests/stream_states.rs

@@ -372,20 +372,15 @@ fn recv_next_stream_id_updated_by_malformed_headers() {
         .recv_frame(frames::go_away(1).protocol_error())
         .close();
 
-        let srv = server::handshake(io)
+    let srv = server::handshake(io)
-            .expect("handshake")
+        .expect("handshake")
-            .and_then(|srv| srv.into_future().then(|res| {
+        .and_then(|srv| srv.into_future().then(|res| {
-                let (err, _) = res.unwrap_err();
+            let (err, _) = res.unwrap_err();
-                assert_eq!(
+            assert_eq!(err.reason(), Some(h2::Reason::PROTOCOL_ERROR));
-                    err.to_string(),
+            Ok::<(), ()>(())
-                    "protocol error: unspecific protocol error detected"
+        }));
-                );
 
-                Ok::<(), ()>(())
+    srv.join(client).wait().expect("wait");
-            })
-        );
-
-        srv.join(client).wait().expect("wait");
 }
 
 #[test]
@@ -403,37 +398,28 @@ fn skipped_stream_ids_are_implicitly_closed() {
         )
         // send the response on a lower-numbered stream, which should be
         // implicitly closed.
-        .send_frame(frames::headers(3).response(200));
+        .send_frame(frames::headers(3).response(299))
+        // however, our client choose to send a RST_STREAM because it
+        // can't tell if it had previously reset '3'.
+        .recv_frame(frames::reset(3).stream_closed())
+        .send_frame(frames::headers(5).response(200).eos());
 
-        let h2 = client::Builder::new()
+    let h2 = client::Builder::new()
-            .initial_stream_id(5)
+        .initial_stream_id(5)
-            .handshake::<_, Bytes>(io)
+        .handshake::<_, Bytes>(io)
-            .expect("handshake")
+        .expect("handshake")
-            .and_then(|(mut client, h2)| {
+        .and_then(|(mut client, h2)| {
-                let req = client
+            let req = client
-                    .get("https://example.com/")
+                .get("https://example.com/")
-                    .then(|res| {
+                .expect("response")
-                        let err = res.unwrap_err();
+                .map(|res| {
-                        assert_eq!(
+                    assert_eq!(res.status(), StatusCode::OK);
-                            err.to_string(),
+                });
-                            "protocol error: unspecific protocol error detected");
+            h2.drive(req)
-                        Ok::<(), ()>(())
+                .and_then(|(conn, ())| conn.expect("client"))
-                    });
+        });
-                    // client should see a conn error
-                    let conn = h2.then(|res| {
-                        let err = res.unwrap_err();
-                        assert_eq!(
-                            err.to_string(),
-                            "protocol error: unspecific protocol error detected"
-                        );
-                        Ok::<(), ()>(())
-                    });
-                    conn.unwrap().join(req)
-            });
-
-
-        h2.join(srv).wait().expect("wait");
 
+    h2.join(srv).wait().expect("wait");
 }
 
 #[test]
@@ -545,7 +531,10 @@ fn rst_stream_expires() {
         .ping_pong([1; 8])
         // sending frame after canceled!
         .send_frame(frames::data(1, vec![0; 16_384]).eos())
-        .recv_frame(frames::go_away(0).protocol_error())
+        // window capacity is returned
+        .recv_frame(frames::window_update(0, 16_384 * 2))
+        // and then stream error
+        .recv_frame(frames::reset(1).stream_closed())
         .close();
 
     let client = client::Builder::new()
@@ -555,23 +544,16 @@ fn rst_stream_expires() {
         .and_then(|(mut client, conn)| {
             let req = client
                 .get("https://example.com/")
+                .expect("response")
                 .map(|resp| {
                     assert_eq!(resp.status(), StatusCode::OK);
                     // drop resp will send a reset
-                })
-                .map_err(|e| -> Error {
-                    unreachable!("req shouldn't error: {:?}", e)
                 });
 
-            conn.drive(req)
+            // no connection error should happen
-                .and_then(|(conn, _)| conn.expect_err("client should error"))
+            conn.expect("client")
-                .map(|err| {
+                .drive(req)
-                    assert_eq!(
+                .and_then(move |(conn, _)| conn.map(move |()| drop(client)))
-                        err.to_string(),
-                        "protocol error: unspecific protocol error detected"
-                    );
-                    drop(client);
-                })
         });
 
     client.join(srv).wait().expect("wait");
@@ -607,9 +589,9 @@ fn rst_stream_max() {
         .send_frame(frames::data(3, vec![0; 16]).eos())
         // ping pong to be sure of no goaway
         .ping_pong([1; 8])
-        // 1 has been evicted, will get a goaway
+        // 1 has been evicted, will get a reset
         .send_frame(frames::data(1, vec![0; 16]).eos())
-        .recv_frame(frames::go_away(0).protocol_error())
+        .recv_frame(frames::reset(1).stream_closed())
         .close();
 
     let client = client::Builder::new()
@@ -619,33 +601,24 @@ fn rst_stream_max() {
         .and_then(|(mut client, conn)| {
             let req1 = client
                 .get("https://example.com/")
+                .expect("response1")
                 .map(|resp| {
                     assert_eq!(resp.status(), StatusCode::OK);
                     // drop resp will send a reset
-                })
-                .map_err(|e| -> Error {
-                    unreachable!("req1 shouldn't error: {:?}", e)
                 });
 
             let req2 = client
                 .get("https://example.com/")
+                .expect("response2")
                 .map(|resp| {
                     assert_eq!(resp.status(), StatusCode::OK);
                     // drop resp will send a reset
-                })
-                .map_err(|e| -> Error {
-                    unreachable!("req2 shouldn't error: {:?}", e)
                 });
 
-            conn.drive(req1.join(req2))
+            // no connection error should happen
-                .and_then(|(conn, _)| conn.expect_err("client"))
+            conn.expect("client")
-                .map(move |err| {
+                .drive(req1.join(req2))
-                    drop(client);
+                .and_then(move |(conn, _)| conn.map(move |()| drop(client)))
-                    assert_eq!(
-                        err.to_string(),
-                        "protocol error: unspecific protocol error detected"
-                    );
-                })
         });