Be more lenient with streams in the pending_send queue. (#261)

The `is_peer_reset()` check doesn't quite cover all the cases where we call
`clear_queue`, such as when we call `recv_err`. Instead of trying to make the
check more precise, let's gracefully handle spurious entries in the queue.

src/proto/streams/prioritize.rs

@@ -647,15 +647,6 @@ impl Prioritize {
                     // To be safe, we just always ask the stream.
                     let is_pending_reset = stream.is_pending_reset_expiration();
 
-                    // If the stream receives a RESET from the peer, it may have
-                    // had data buffered to be sent, but all the frames are cleared
-                    // in clear_queue(). Instead of doing O(N) traversal through queue
-                    // to remove, lets just ignore peer_reset streams here.
-                    if stream.state.is_peer_reset() {
-                        counts.transition_after(stream, is_pending_reset);
-                        continue;
-                    }
-
                     trace!(" --> stream={:?}; is_pending_reset={:?};",
                         stream.id, is_pending_reset);
 
@@ -757,13 +748,23 @@ impl Prioritize {
                              )
                         ),
                         None => {
-                            let reason = stream.state.get_scheduled_reset()
+                            if let Some(reason) = stream.state.get_scheduled_reset() {
-                                .expect("must be scheduled to reset");
-
                                 stream.state.set_reset(reason);
 
                                 let frame = frame::Reset::new(stream.id, reason);
                                 Frame::Reset(frame)
+                            } else {
+                                // If the stream receives a RESET from the peer, it may have
+                                // had data buffered to be sent, but all the frames are cleared
+                                // in clear_queue(). Instead of doing O(N) traversal through queue
+                                // to remove, lets just ignore the stream here.
+                                trace!("removing dangling stream from pending_send");
+                                // Since this should only happen as a consequence of `clear_queue`,
+                                // we must be in a closed state of some kind.
+                                debug_assert!(stream.state.is_closed());
+                                counts.transition_after(stream, is_pending_reset);
+                                continue;
+                            }
                         }
                     };
 

src/proto/streams/state.rs

@@ -325,13 +325,6 @@ impl State {
         }
     }
 
-    pub fn is_peer_reset(&self) -> bool {
-        match self.inner {
-            Closed(Cause::Proto(_)) => true,
-            _ => false,
-        }
-    }
-
     /// Returns true if the stream is already reset.
     pub fn is_reset(&self) -> bool {
         match self.inner {