Reset any queued stream on receipt of remote reset

src/proto/streams/prioritize.rs

@@ -593,6 +593,8 @@ impl Prioritize {
                     // To be safe, we just always ask the stream.
                     let is_counted = stream.is_counted();
                     let is_pending_reset = stream.is_pending_reset_expiration();
+                    trace!(" --> stream={:?}; is_pending_reset={:?};",
+                        stream.id, is_pending_reset);
 
                     let frame = match stream.pending_send.pop_front(buffer) {
                         Some(Frame::Data(mut frame)) => {
@@ -603,13 +605,14 @@ impl Prioritize {
 
                             trace!(
                                 " --> data frame; stream={:?}; sz={}; eos={:?}; window={}; \
-                                 available={}; requested={}",
+                                 available={}; requested={}; buffered={};",
                                 frame.stream_id(),
                                 sz,
                                 frame.is_end_stream(),
                                 stream_capacity,
                                 stream.send_flow.available(),
-                                stream.requested_send_capacity
+                                stream.requested_send_capacity,
+                                stream.buffered_send_data,
                             );
 
                             // Zero length data frames always have capacity to

src/proto/streams/state.rs

@@ -215,19 +215,32 @@ impl State {
 
     /// The remote explicitly sent a RST_STREAM.
     pub fn recv_reset(&mut self, reason: Reason, queued: bool) {
-
         match self.inner {
-            Closed(Cause::EndStream) if queued => {
-                // If the stream has a queued EOS frame, transition to peer
-                // reset.
-                trace!("recv_reset: reason={:?}; queued=true", reason);
-                self.inner = Closed(Cause::Proto(reason));
-            },
-            Closed(..) => {},
-            _ => {
-                trace!("recv_reset; reason={:?}", reason);
+            // If the stream is already in a `Closed` state, do nothing,
+            // provided that there are no frames still in the send queue.
+            Closed(..) if !queued => {},
+            // A notionally `Closed` stream may still have queued frames in
+            // the following cases:
+            //
+            // - if the cause is `Cause::Scheduled(..)` (i.e. we have not
+            //   actually closed the stream yet).
+            // - if the cause is `Cause::EndStream`: we transition to this
+            //   state when an EOS frame is *enqueued* (so that it's invalid
+            //   to enqueue more frames), not when the EOS frame is *sent*;
+            //   therefore, there may still be frames ahead of the EOS frame
+            //   in the send queue.
+            //
+            // In either of these cases, we want to honor the received
+            // RST_STREAM by resetting the stream immediately (and clearing
+            // the send queue).
+            state => {
+                trace!(
+                    "recv_reset; reason={:?}; state={:?}; queued={:?}",
+                    reason, state, queued
+                );
                 self.inner = Closed(Cause::Proto(reason));
             },
+
         }
     }