Repository: hyperium/h2 (rust)

Issue:
Handle `Stream` dropped before end-of-stream flag sent
This is a continuation of [a comment](https://github.com/carllerche/h2/pull/109/files#r143109861).

I believe that currently, if a `Stream` is dropped without sending an end-of-stream flag, nothing happens. This would cause the stream to enter a hung state.


Discussion:
We have to guess at the intent of the user here. 

- If there was a content-length set, and it's gotten to zero, then that's easy: send a 0-length EOS frame.
- If there was a content-length set, and it's not at zero yet, then the stream should be reset (probably with a CANCEL code).
- If no content-length was set, then I'm not sure which is correct. Do we guess that the user is just done? Or is it more appropriate to assume the stream should be reset unless explicitly finished?
- With the client, it probably shouldn't reset unless the user also dropped the response future/stream, in which case dropping all stream refs triggers a cancel anyways.
- With the server, I'm still not certain whether it's more correct to send a reset or an empty EOS data frame.
IMO for something like h2, I would be happy to say that the lib shouldn't maintain additional state to guess the right action here. Instead, it is up to the user to include an EOS flag at the end of the stream. Higher level libs should hide this detail from their users.
So, unequivocally send a reset of `CANCEL`, always?
I would say that would be preferable to hanging the stream and I don't think we should track additional state. It may be that the library already has enough state to do the right thing, but I am not sure off the top of my head.

So, in short:

* Don't track additional state for this issue
* Do the best thing possible, even if that is always sending a RST_STREAM with CANCEL.

Patch:
diff --git a/src/proto/streams/prioritize.rs b/src/proto/streams/prioritize.rs
--- a/src/proto/streams/prioritize.rs
+++ b/src/proto/streams/prioritize.rs
@@ -266,6 +266,27 @@ impl Prioritize {
         Ok(())
     }
 
+    /// Reclaim all capacity assigned to the stream and re-assign it to the
+    /// connection
+    pub fn reclaim_all_capacity(&mut self, stream: &mut store::Ptr) {
+        let available = stream.send_flow.available().as_size();
+        stream.send_flow.claim_capacity(available);
+        // Re-assign all capacity to the connection
+        self.assign_connection_capacity(available, stream);
+    }
+
+    /// Reclaim just reserved capacity, not buffered capacity, and re-assign
+    /// it to the connection
+    pub fn reclaim_reserved_capacity(&mut self, stream: &mut store::Ptr) {
+        // only reclaim requested capacity that isn't already buffered
+        if stream.requested_send_capacity > stream.buffered_send_data {
+            let reserved = stream.requested_send_capacity - stream.buffered_send_data;
+
+            stream.send_flow.claim_capacity(reserved);
+            self.assign_connection_capacity(reserved, stream);
+        }
+    }
+
     pub fn assign_connection_capacity<R>(&mut self, inc: WindowSize, store: &mut R)
     where
         R: Resolve,
diff --git a/src/proto/streams/send.rs b/src/proto/streams/send.rs
--- a/src/proto/streams/send.rs
+++ b/src/proto/streams/send.rs
@@ -155,6 +155,7 @@ impl Send {
     }
 
     pub fn schedule_cancel(&mut self, stream: &mut store::Ptr, task: &mut Option<Task>) {
+        trace!("schedule_cancel; {:?}", stream.id);
         if stream.state.is_closed() {
             // Stream is already closed, nothing more to do
             return;
@@ -162,7 +163,7 @@ impl Send {
 
         stream.state.set_canceled();
 
-        self.reclaim_capacity(stream);
+        self.prioritize.reclaim_reserved_capacity(stream);
         self.prioritize.schedule_send(stream, task);
     }
 
@@ -285,17 +286,7 @@ impl Send {
     ) {
         // Clear all pending outbound frames
         self.prioritize.clear_queue(buffer, stream);
-        self.reclaim_capacity(stream);
-    }
-
-    fn reclaim_capacity(&mut self, stream: &mut store::Ptr) {
-        // Reclaim all capacity assigned to the stream and re-assign it to the
-        // connection
-        let available = stream.send_flow.available().as_size();
-        stream.send_flow.claim_capacity(available);
-        // Re-assign all capacity to the connection
-        self.prioritize
-            .assign_connection_capacity(available, stream);
+        self.prioritize.reclaim_all_capacity(stream);
     }
 
     pub fn apply_remote_settings<B>(
diff --git a/src/proto/streams/stream.rs b/src/proto/streams/stream.rs
--- a/src/proto/streams/stream.rs
+++ b/src/proto/streams/stream.rs
@@ -243,7 +243,7 @@ impl Stream {
     ///
     /// In this case, a reset should be sent.
     pub fn is_canceled_interest(&self) -> bool {
-        self.ref_count == 0 && !self.state.is_recv_closed()
+        self.ref_count == 0 && !self.state.is_closed()
     }
 
     pub fn assign_capacity(&mut self, capacity: WindowSize) {


Test patch:
diff --git a/tests/server.rs b/tests/server.rs
--- a/tests/server.rs
+++ b/tests/server.rs
@@ -172,7 +172,7 @@ fn recv_connection_header() {
 }
 
 #[test]
-fn sends_reset_cancel_when_body_is_dropped() {
+fn sends_reset_cancel_when_req_body_is_dropped() {
     let _ = ::env_logger::init();
     let (io, client) = mock::new();
 
@@ -203,3 +203,61 @@ fn sends_reset_cancel_when_body_is_dropped() {
 
     srv.join(client).wait().expect("wait");
 }
+
+#[test]
+fn sends_reset_cancel_when_res_body_is_dropped() {
+    let _ = ::env_logger::init();
+    let (io, client) = mock::new();
+
+    let client = client
+        .assert_server_handshake()
+        .unwrap()
+        .recv_settings()
+        .send_frame(
+            frames::headers(1)
+                .request("GET", "https://example.com/")
+                .eos()
+        )
+        .recv_frame(frames::headers(1).response(200))
+        .recv_frame(frames::reset(1).cancel())
+        .send_frame(
+            frames::headers(3)
+                .request("GET", "https://example.com/")
+                .eos()
+        )
+        .recv_frame(frames::headers(3).response(200))
+        .recv_frame(frames::data(3, vec![0; 10]))
+        .recv_frame(frames::reset(3).cancel())
+        .close();
+
+    let srv = Server::handshake(io).expect("handshake").and_then(|srv| {
+        srv.into_future().unwrap().and_then(|(reqstream, srv)| {
+            let (req, mut stream) = reqstream.unwrap();
+
+            assert_eq!(req.method(), &http::Method::GET);
+
+            let rsp = http::Response::builder()
+                .status(200)
+                .body(())
+                .unwrap();
+            stream.send_response(rsp, false).unwrap();
+            // SendStream dropped
+
+            srv.into_future().unwrap()
+        }).and_then(|(reqstream, srv)| {
+            let (_req, mut stream) = reqstream.unwrap();
+
+            let rsp = http::Response::builder()
+                .status(200)
+                .body(())
+                .unwrap();
+            let mut tx = stream.send_response(rsp, false).unwrap();
+            tx.send_data(vec![0; 10].into(), false).unwrap();
+            // no send_data with eos
+
+            srv.into_future().unwrap()
+        })
+    });
+
+    srv.join(client).wait().expect("wait");
+}