Repository: hyperium/h2 (rust)

Issue:
Receiving a SETTINGS frame that reduces window size causes queued streams to panic
This test panics. This may be resolved by #79

```rust
#[test]
fn something_funky() {
    let _ = ::env_logger::init();
    let (io, srv) = mock::new();

    let h2 = Client::handshake(io).unwrap()
        .and_then(|mut h2| {
            let request = Request::builder()
                .method(Method::POST)
                .uri("https://http2.akamai.com/")
                .body(()).unwrap();

            let mut stream = h2.request(request, false).unwrap();

            stream.reserve_capacity(11);

            h2.drive(util::wait_for_capacity(stream, 11))
        })
        .and_then(|(h2, mut stream)| {
            assert_eq!(stream.capacity(), 11);

            stream.send_data("hello world".into(), true).unwrap();

            h2.drive(GetResponse { stream: Some(stream) })
        })
        .and_then(|(h2, (response, _))| {
            assert_eq!(response.status(), StatusCode::NO_CONTENT);

            // Wait for the connection to close
            h2.unwrap()
        })
        ;

    let mut settings = frame::Settings::default();
    settings.set_initial_window_size(0);

    let srv = srv.assert_client_handshake_with_settings(settings).unwrap()
        .recv_settings()
        .recv_frame(
            frames::headers(1)
                .request("POST", "https://http2.akamai.com/")
        )
        .idle_ms(100)
        .send_frame(
            frames::window_update(0, 11))
        .send_frame(
            frames::window_update(1, 11))
        .recv_frame(
            frames::data(1, "hello world").eos())
        .send_frame(
            frames::headers(1)
                .response(204)
                .eos()
        )
        /*
        .recv_frame(frames::data(1, "hello").eos())
        .send_frame(frames::window_update(1, 5))
        */
        .map(drop)
        ;

    let _ = h2.join(srv)
        .wait().unwrap();
}
```


Discussion:

Patch:
diff --git a/src/proto/streams/flow_control.rs b/src/proto/streams/flow_control.rs
--- a/src/proto/streams/flow_control.rs
+++ b/src/proto/streams/flow_control.rs
@@ -126,8 +126,10 @@ impl FlowControl {
     /// This is called after receiving a SETTINGS frame with a lower
     /// INITIAL_WINDOW_SIZE value.
     pub fn dec_window(&mut self, sz: WindowSize) {
+        trace!("dec_window; sz={}; window={}, available={}", sz, self.window_size, self.available);
         // This should not be able to overflow `window_size` from the bottom.
         self.window_size -= sz as i32;
+        self.available = self.available.saturating_sub(sz);
     }
 
     /// Decrements the window reflecting data has actually been sent. The caller
diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs
--- a/src/proto/streams/streams.rs
+++ b/src/proto/streams/streams.rs
@@ -734,12 +734,20 @@ where
     P: Peer,
 {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        let me = self.inner.lock().unwrap();
-        let stream = &me.store[self.key];
-        fmt.debug_struct("StreamRef")
-            .field("stream_id", &stream.id)
-            .field("ref_count", &stream.ref_count)
-            .finish()
+        match self.inner.lock() {
+            Ok(me) => {
+                let stream = &me.store[self.key];
+                fmt.debug_struct("StreamRef")
+                    .field("stream_id", &stream.id)
+                    .field("ref_count", &stream.ref_count)
+                    .finish()
+            },
+            Err(_poisoned) => {
+                fmt.debug_struct("StreamRef")
+                    .field("inner", &"<Poisoned>")
+                    .finish()
+            }
+        }
     }
 }
 


Test patch:
diff --git a/tests/flow_control.rs b/tests/flow_control.rs
--- a/tests/flow_control.rs
+++ b/tests/flow_control.rs
@@ -532,17 +532,16 @@ fn recv_window_update_on_stream_closed_by_data_frame() {
             // Send a data frame, this will also close the connection
             stream.send_data("hello".into(), true).unwrap();
 
+            // keep `stream` from being dropped in order to prevent
+            // it from sending an RST_STREAM frame.
+            //
+            // i know this is kind of evil, but it's necessary to
+            // ensure that the stream is closed by the EOS frame,
+            // and not by the RST_STREAM.
+            std::mem::forget(stream);
+
             // Wait for the connection to close
-            h2.map(|h2| {
-                // keep `stream` from being dropped in order to prevent
-                // it from sending an RST_STREAM frame.
-                std::mem::forget(stream);
-                // i know this is kind of evil, but it's necessary to
-                // ensure that the stream is closed by the EOS frame,
-                // and not by the RST_STREAM.
-                h2
-            })
-                .unwrap()
+            h2.unwrap()
         });
 
     let srv = srv.assert_client_handshake()
@@ -744,3 +743,60 @@ fn connection_notified_on_released_capacity() {
     // implicitly released before.
     drop(b);
 }
+
+#[test]
+fn recv_settings_removes_available_capacity() {
+    let _ = ::env_logger::init();
+    let (io, srv) = mock::new();
+
+    let mut settings = frame::Settings::default();
+    settings.set_initial_window_size(Some(0));
+
+    let srv = srv.assert_client_handshake_with_settings(settings).unwrap()
+        .recv_settings()
+        .recv_frame(
+            frames::headers(1)
+                .request("POST", "https://http2.akamai.com/")
+        )
+        .idle_ms(100)
+        .send_frame(frames::window_update(0, 11))
+        .send_frame(frames::window_update(1, 11))
+        .recv_frame(frames::data(1, "hello world").eos())
+        .send_frame(
+            frames::headers(1)
+                .response(204)
+                .eos()
+        )
+        .close();
+
+
+    let h2 = Client::handshake(io).unwrap()
+        .and_then(|(mut client, h2)| {
+            let request = Request::builder()
+                .method(Method::POST)
+                .uri("https://http2.akamai.com/")
+                .body(()).unwrap();
+
+            let mut stream = client.send_request(request, false).unwrap();
+
+            stream.reserve_capacity(11);
+
+            h2.drive(util::wait_for_capacity(stream, 11))
+        })
+        .and_then(|(h2, mut stream)| {
+            assert_eq!(stream.capacity(), 11);
+
+            stream.send_data("hello world".into(), true).unwrap();
+
+            h2.drive(GetResponse { stream: Some(stream) })
+        })
+        .and_then(|(h2, (response, _))| {
+            assert_eq!(response.status(), StatusCode::NO_CONTENT);
+
+            // Wait for the connection to close
+            h2.unwrap()
+        });
+
+    let _ = h2.join(srv)
+        .wait().unwrap();
+}