Adding upstream version 110.0.1.upstream/110.0.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1576 insertions, 0 deletions

diff --git a/third_party/rust/h2/src/proto/streams/streams.rs b/third_party/rust/h2/src/proto/streams/streams.rs
new file mode 100644
index 0000000000..aee64ca6a3
--- /dev/null
+++ b/third_party/rust/h2/src/proto/streams/streams.rs
@@ -0,0 +1,1576 @@
+use super::recv::RecvHeaderBlockError;
+use super::store::{self, Entry, Resolve, Store};
+use super::{Buffer, Config, Counts, Prioritized, Recv, Send, Stream, StreamId};
+use crate::codec::{Codec, SendError, UserError};
+use crate::ext::Protocol;
+use crate::frame::{self, Frame, Reason};
+use crate::proto::{peer, Error, Initiator, Open, Peer, WindowSize};
+use crate::{client, proto, server};
+
+use bytes::{Buf, Bytes};
+use http::{HeaderMap, Request, Response};
+use std::task::{Context, Poll, Waker};
+use tokio::io::AsyncWrite;
+
+use std::sync::{Arc, Mutex};
+use std::{fmt, io};
+
+#[derive(Debug)]
+pub(crate) struct Streams<B, P>
+where
+    P: Peer,
+{
+    /// Holds most of the connection and stream related state for processing
+    /// HTTP/2 frames associated with streams.
+    inner: Arc<Mutex<Inner>>,
+
+    /// This is the queue of frames to be written to the wire. This is split out
+    /// to avoid requiring a `B` generic on all public API types even if `B` is
+    /// not technically required.
+    ///
+    /// Currently, splitting this out requires a second `Arc` + `Mutex`.
+    /// However, it should be possible to avoid this duplication with a little
+    /// bit of unsafe code. This optimization has been postponed until it has
+    /// been shown to be necessary.
+    send_buffer: Arc<SendBuffer<B>>,
+
+    _p: ::std::marker::PhantomData<P>,
+}
+
+// Like `Streams` but with a `peer::Dyn` field instead of a static `P: Peer` type parameter.
+// Ensures that the methods only get one instantiation, instead of two (client and server)
+#[derive(Debug)]
+pub(crate) struct DynStreams<'a, B> {
+    inner: &'a Mutex<Inner>,
+
+    send_buffer: &'a SendBuffer<B>,
+
+    peer: peer::Dyn,
+}
+
+/// Reference to the stream state
+#[derive(Debug)]
+pub(crate) struct StreamRef<B> {
+    opaque: OpaqueStreamRef,
+    send_buffer: Arc<SendBuffer<B>>,
+}
+
+/// Reference to the stream state that hides the send data chunk generic
+pub(crate) struct OpaqueStreamRef {
+    inner: Arc<Mutex<Inner>>,
+    key: store::Key,
+}
+
+/// Fields needed to manage state related to managing the set of streams. This
+/// is mostly split out to make ownership happy.
+///
+/// TODO: better name
+#[derive(Debug)]
+struct Inner {
+    /// Tracks send & recv stream concurrency.
+    counts: Counts,
+
+    /// Connection level state and performs actions on streams
+    actions: Actions,
+
+    /// Stores stream state
+    store: Store,
+
+    /// The number of stream refs to this shared state.
+    refs: usize,
+}
+
+#[derive(Debug)]
+struct Actions {
+    /// Manages state transitions initiated by receiving frames
+    recv: Recv,
+
+    /// Manages state transitions initiated by sending frames
+    send: Send,
+
+    /// Task that calls `poll_complete`.
+    task: Option<Waker>,
+
+    /// If the connection errors, a copy is kept for any StreamRefs.
+    conn_error: Option<proto::Error>,
+}
+
+/// Contains the buffer of frames to be written to the wire.
+#[derive(Debug)]
+struct SendBuffer<B> {
+    inner: Mutex<Buffer<Frame<B>>>,
+}
+
+// ===== impl Streams =====
+
+impl<B, P> Streams<B, P>
+where
+    B: Buf,
+    P: Peer,
+{
+    pub fn new(config: Config) -> Self {
+        let peer = P::r#dyn();
+
+        Streams {
+            inner: Inner::new(peer, config),
+            send_buffer: Arc::new(SendBuffer::new()),
+            _p: ::std::marker::PhantomData,
+        }
+    }
+
+    pub fn set_target_connection_window_size(&mut self, size: WindowSize) {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        me.actions
+            .recv
+            .set_target_connection_window(size, &mut me.actions.task)
+    }
+
+    pub fn next_incoming(&mut self) -> Option<StreamRef<B>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+        me.actions.recv.next_incoming(&mut me.store).map(|key| {
+            let stream = &mut me.store.resolve(key);
+            tracing::trace!(
+                "next_incoming; id={:?}, state={:?}",
+                stream.id,
+                stream.state
+            );
+            // TODO: ideally, OpaqueStreamRefs::new would do this, but we're holding
+            // the lock, so it can't.
+            me.refs += 1;
+            StreamRef {
+                opaque: OpaqueStreamRef::new(self.inner.clone(), stream),
+                send_buffer: self.send_buffer.clone(),
+            }
+        })
+    }
+
+    pub fn send_pending_refusal<T>(
+        &mut self,
+        cx: &mut Context,
+        dst: &mut Codec<T, Prioritized<B>>,
+    ) -> Poll<io::Result<()>>
+    where
+        T: AsyncWrite + Unpin,
+    {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+        me.actions.recv.send_pending_refusal(cx, dst)
+    }
+
+    pub fn clear_expired_reset_streams(&mut self) {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+        me.actions
+            .recv
+            .clear_expired_reset_streams(&mut me.store, &mut me.counts);
+    }
+
+    pub fn poll_complete<T>(
+        &mut self,
+        cx: &mut Context,
+        dst: &mut Codec<T, Prioritized<B>>,
+    ) -> Poll<io::Result<()>>
+    where
+        T: AsyncWrite + Unpin,
+    {
+        let mut me = self.inner.lock().unwrap();
+        me.poll_complete(&self.send_buffer, cx, dst)
+    }
+
+    pub fn apply_remote_settings(&mut self, frame: &frame::Settings) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.counts.apply_remote_settings(frame);
+
+        me.actions.send.apply_remote_settings(
+            frame,
+            send_buffer,
+            &mut me.store,
+            &mut me.counts,
+            &mut me.actions.task,
+        )
+    }
+
+    pub fn apply_local_settings(&mut self, frame: &frame::Settings) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        me.actions.recv.apply_local_settings(frame, &mut me.store)
+    }
+
+    pub fn send_request(
+        &mut self,
+        mut request: Request<()>,
+        end_of_stream: bool,
+        pending: Option<&OpaqueStreamRef>,
+    ) -> Result<StreamRef<B>, SendError> {
+        use super::stream::ContentLength;
+        use http::Method;
+
+        let protocol = request.extensions_mut().remove::<Protocol>();
+
+        // Clear before taking lock, incase extensions contain a StreamRef.
+        request.extensions_mut().clear();
+
+        // TODO: There is a hazard with assigning a stream ID before the
+        // prioritize layer. If prioritization reorders new streams, this
+        // implicitly closes the earlier stream IDs.
+        //
+        // See: hyperium/h2#11
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.actions.ensure_no_conn_error()?;
+        me.actions.send.ensure_next_stream_id()?;
+
+        // The `pending` argument is provided by the `Client`, and holds
+        // a store `Key` of a `Stream` that may have been not been opened
+        // yet.
+        //
+        // If that stream is still pending, the Client isn't allowed to
+        // queue up another pending stream. They should use `poll_ready`.
+        if let Some(stream) = pending {
+            if me.store.resolve(stream.key).is_pending_open {
+                return Err(UserError::Rejected.into());
+            }
+        }
+
+        if me.counts.peer().is_server() {
+            // Servers cannot open streams. PushPromise must first be reserved.
+            return Err(UserError::UnexpectedFrameType.into());
+        }
+
+        let stream_id = me.actions.send.open()?;
+
+        let mut stream = Stream::new(
+            stream_id,
+            me.actions.send.init_window_sz(),
+            me.actions.recv.init_window_sz(),
+        );
+
+        if *request.method() == Method::HEAD {
+            stream.content_length = ContentLength::Head;
+        }
+
+        // Convert the message
+        let headers =
+            client::Peer::convert_send_message(stream_id, request, protocol, end_of_stream)?;
+
+        let mut stream = me.store.insert(stream.id, stream);
+
+        let sent = me.actions.send.send_headers(
+            headers,
+            send_buffer,
+            &mut stream,
+            &mut me.counts,
+            &mut me.actions.task,
+        );
+
+        // send_headers can return a UserError, if it does,
+        // we should forget about this stream.
+        if let Err(err) = sent {
+            stream.unlink();
+            stream.remove();
+            return Err(err.into());
+        }
+
+        // Given that the stream has been initialized, it should not be in the
+        // closed state.
+        debug_assert!(!stream.state.is_closed());
+
+        // TODO: ideally, OpaqueStreamRefs::new would do this, but we're holding
+        // the lock, so it can't.
+        me.refs += 1;
+
+        Ok(StreamRef {
+            opaque: OpaqueStreamRef::new(self.inner.clone(), &mut stream),
+            send_buffer: self.send_buffer.clone(),
+        })
+    }
+
+    pub(crate) fn is_extended_connect_protocol_enabled(&self) -> bool {
+        self.inner
+            .lock()
+            .unwrap()
+            .actions
+            .send
+            .is_extended_connect_protocol_enabled()
+    }
+}
+
+impl<B> DynStreams<'_, B> {
+    pub fn recv_headers(&mut self, frame: frame::Headers) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+
+        me.recv_headers(self.peer, &self.send_buffer, frame)
+    }
+
+    pub fn recv_data(&mut self, frame: frame::Data) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        me.recv_data(self.peer, &self.send_buffer, frame)
+    }
+
+    pub fn recv_reset(&mut self, frame: frame::Reset) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+
+        me.recv_reset(&self.send_buffer, frame)
+    }
+
+    /// Notify all streams that a connection-level error happened.
+    pub fn handle_error(&mut self, err: proto::Error) -> StreamId {
+        let mut me = self.inner.lock().unwrap();
+        me.handle_error(&self.send_buffer, err)
+    }
+
+    pub fn recv_go_away(&mut self, frame: &frame::GoAway) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        me.recv_go_away(&self.send_buffer, frame)
+    }
+
+    pub fn last_processed_id(&self) -> StreamId {
+        self.inner.lock().unwrap().actions.recv.last_processed_id()
+    }
+
+    pub fn recv_window_update(&mut self, frame: frame::WindowUpdate) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        me.recv_window_update(&self.send_buffer, frame)
+    }
+
+    pub fn recv_push_promise(&mut self, frame: frame::PushPromise) -> Result<(), Error> {
+        let mut me = self.inner.lock().unwrap();
+        me.recv_push_promise(&self.send_buffer, frame)
+    }
+
+    pub fn recv_eof(&mut self, clear_pending_accept: bool) -> Result<(), ()> {
+        let mut me = self.inner.lock().map_err(|_| ())?;
+        me.recv_eof(&self.send_buffer, clear_pending_accept)
+    }
+
+    pub fn send_reset(&mut self, id: StreamId, reason: Reason) {
+        let mut me = self.inner.lock().unwrap();
+        me.send_reset(&self.send_buffer, id, reason)
+    }
+
+    pub fn send_go_away(&mut self, last_processed_id: StreamId) {
+        let mut me = self.inner.lock().unwrap();
+        me.actions.recv.go_away(last_processed_id);
+    }
+}
+
+impl Inner {
+    fn new(peer: peer::Dyn, config: Config) -> Arc<Mutex<Self>> {
+        Arc::new(Mutex::new(Inner {
+            counts: Counts::new(peer, &config),
+            actions: Actions {
+                recv: Recv::new(peer, &config),
+                send: Send::new(&config),
+                task: None,
+                conn_error: None,
+            },
+            store: Store::new(),
+            refs: 1,
+        }))
+    }
+
+    fn recv_headers<B>(
+        &mut self,
+        peer: peer::Dyn,
+        send_buffer: &SendBuffer<B>,
+        frame: frame::Headers,
+    ) -> Result<(), Error> {
+        let id = frame.stream_id();
+
+        // The GOAWAY process has begun. All streams with a greater ID than
+        // specified as part of GOAWAY should be ignored.
+        if id > self.actions.recv.max_stream_id() {
+            tracing::trace!(
+                "id ({:?}) > max_stream_id ({:?}), ignoring HEADERS",
+                id,
+                self.actions.recv.max_stream_id()
+            );
+            return Ok(());
+        }
+
+        let key = match self.store.find_entry(id) {
+            Entry::Occupied(e) => e.key(),
+            Entry::Vacant(e) => {
+                // Client: it's possible to send a request, and then send
+                // a RST_STREAM while the response HEADERS were in transit.
+                //
+                // Server: we can't reset a stream before having received
+                // the request headers, so don't allow.
+                if !peer.is_server() {
+                    // This may be response headers for a stream we've already
+                    // forgotten about...
+                    if self.actions.may_have_forgotten_stream(peer, id) {
+                        tracing::debug!(
+                            "recv_headers for old stream={:?}, sending STREAM_CLOSED",
+                            id,
+                        );
+                        return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
+                    }
+                }
+
+                match self
+                    .actions
+                    .recv
+                    .open(id, Open::Headers, &mut self.counts)?
+                {
+                    Some(stream_id) => {
+                        let stream = Stream::new(
+                            stream_id,
+                            self.actions.send.init_window_sz(),
+                            self.actions.recv.init_window_sz(),
+                        );
+
+                        e.insert(stream)
+                    }
+                    None => return Ok(()),
+                }
+            }
+        };
+
+        let stream = self.store.resolve(key);
+
+        if stream.state.is_local_reset() {
+            // Locally reset streams must ignore frames "for some time".
+            // This is because the remote may have sent trailers before
+            // receiving the RST_STREAM frame.
+            tracing::trace!("recv_headers; ignoring trailers on {:?}", stream.id);
+            return Ok(());
+        }
+
+        let actions = &mut self.actions;
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        self.counts.transition(stream, |counts, stream| {
+            tracing::trace!(
+                "recv_headers; stream={:?}; state={:?}",
+                stream.id,
+                stream.state
+            );
+
+            let res = if stream.state.is_recv_headers() {
+                match actions.recv.recv_headers(frame, stream, counts) {
+                    Ok(()) => Ok(()),
+                    Err(RecvHeaderBlockError::Oversize(resp)) => {
+                        if let Some(resp) = resp {
+                            let sent = actions.send.send_headers(
+                                resp, send_buffer, stream, counts, &mut actions.task);
+                            debug_assert!(sent.is_ok(), "oversize response should not fail");
+
+                            actions.send.schedule_implicit_reset(
+                                stream,
+                                Reason::REFUSED_STREAM,
+                                counts,
+                                &mut actions.task);
+
+                            actions.recv.enqueue_reset_expiration(stream, counts);
+
+                            Ok(())
+                        } else {
+                            Err(Error::library_reset(stream.id, Reason::REFUSED_STREAM))
+                        }
+                    },
+                    Err(RecvHeaderBlockError::State(err)) => Err(err),
+                }
+            } else {
+                if !frame.is_end_stream() {
+                    // Receiving trailers that don't set EOS is a "malformed"
+                    // message. Malformed messages are a stream error.
+                    proto_err!(stream: "recv_headers: trailers frame was not EOS; stream={:?}", stream.id);
+                    return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
+                }
+
+                actions.recv.recv_trailers(frame, stream)
+            };
+
+            actions.reset_on_recv_stream_err(send_buffer, stream, counts, res)
+        })
+    }
+
+    fn recv_data<B>(
+        &mut self,
+        peer: peer::Dyn,
+        send_buffer: &SendBuffer<B>,
+        frame: frame::Data,
+    ) -> Result<(), Error> {
+        let id = frame.stream_id();
+
+        let stream = match self.store.find_mut(&id) {
+            Some(stream) => stream,
+            None => {
+                // The GOAWAY process has begun. All streams with a greater ID
+                // than specified as part of GOAWAY should be ignored.
+                if id > self.actions.recv.max_stream_id() {
+                    tracing::trace!(
+                        "id ({:?}) > max_stream_id ({:?}), ignoring DATA",
+                        id,
+                        self.actions.recv.max_stream_id()
+                    );
+                    return Ok(());
+                }
+
+                if self.actions.may_have_forgotten_stream(peer, id) {
+                    tracing::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;
+
+                    self.actions.recv.ignore_data(sz)?;
+                    return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
+                }
+
+                proto_err!(conn: "recv_data: stream not found; id={:?}", id);
+                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
+            }
+        };
+
+        let actions = &mut self.actions;
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        self.counts.transition(stream, |counts, stream| {
+            let sz = frame.payload().len();
+            let res = actions.recv.recv_data(frame, stream);
+
+            // Any stream error after receiving a DATA frame means
+            // we won't give the data to the user, and so they can't
+            // release the capacity. We do it automatically.
+            if let Err(Error::Reset(..)) = res {
+                actions
+                    .recv
+                    .release_connection_capacity(sz as WindowSize, &mut None);
+            }
+            actions.reset_on_recv_stream_err(send_buffer, stream, counts, res)
+        })
+    }
+
+    fn recv_reset<B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        frame: frame::Reset,
+    ) -> Result<(), Error> {
+        let id = frame.stream_id();
+
+        if id.is_zero() {
+            proto_err!(conn: "recv_reset: invalid stream ID 0");
+            return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
+        }
+
+        // The GOAWAY process has begun. All streams with a greater ID than
+        // specified as part of GOAWAY should be ignored.
+        if id > self.actions.recv.max_stream_id() {
+            tracing::trace!(
+                "id ({:?}) > max_stream_id ({:?}), ignoring RST_STREAM",
+                id,
+                self.actions.recv.max_stream_id()
+            );
+            return Ok(());
+        }
+
+        let stream = match self.store.find_mut(&id) {
+            Some(stream) => stream,
+            None => {
+                // TODO: Are there other error cases?
+                self.actions
+                    .ensure_not_idle(self.counts.peer(), id)
+                    .map_err(Error::library_go_away)?;
+
+                return Ok(());
+            }
+        };
+
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        let actions = &mut self.actions;
+
+        self.counts.transition(stream, |counts, stream| {
+            actions.recv.recv_reset(frame, stream);
+            actions.send.handle_error(send_buffer, stream, counts);
+            assert!(stream.state.is_closed());
+            Ok(())
+        })
+    }
+
+    fn recv_window_update<B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        frame: frame::WindowUpdate,
+    ) -> Result<(), Error> {
+        let id = frame.stream_id();
+
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        if id.is_zero() {
+            self.actions
+                .send
+                .recv_connection_window_update(frame, &mut self.store, &mut self.counts)
+                .map_err(Error::library_go_away)?;
+        } else {
+            // The remote may send window updates for streams that the local now
+            // considers closed. It's ok...
+            if let Some(mut stream) = self.store.find_mut(&id) {
+                // This result is ignored as there is nothing to do when there
+                // is an error. The stream is reset by the function on error and
+                // the error is informational.
+                let _ = self.actions.send.recv_stream_window_update(
+                    frame.size_increment(),
+                    send_buffer,
+                    &mut stream,
+                    &mut self.counts,
+                    &mut self.actions.task,
+                );
+            } else {
+                self.actions
+                    .ensure_not_idle(self.counts.peer(), id)
+                    .map_err(Error::library_go_away)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn handle_error<B>(&mut self, send_buffer: &SendBuffer<B>, err: proto::Error) -> StreamId {
+        let actions = &mut self.actions;
+        let counts = &mut self.counts;
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        let last_processed_id = actions.recv.last_processed_id();
+
+        self.store.for_each(|stream| {
+            counts.transition(stream, |counts, stream| {
+                actions.recv.handle_error(&err, &mut *stream);
+                actions.send.handle_error(send_buffer, stream, counts);
+            })
+        });
+
+        actions.conn_error = Some(err);
+
+        last_processed_id
+    }
+
+    fn recv_go_away<B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        frame: &frame::GoAway,
+    ) -> Result<(), Error> {
+        let actions = &mut self.actions;
+        let counts = &mut self.counts;
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        let last_stream_id = frame.last_stream_id();
+
+        actions.send.recv_go_away(last_stream_id)?;
+
+        let err = Error::remote_go_away(frame.debug_data().clone(), frame.reason());
+
+        self.store.for_each(|stream| {
+            if stream.id > last_stream_id {
+                counts.transition(stream, |counts, stream| {
+                    actions.recv.handle_error(&err, &mut *stream);
+                    actions.send.handle_error(send_buffer, stream, counts);
+                })
+            }
+        });
+
+        actions.conn_error = Some(err);
+
+        Ok(())
+    }
+
+    fn recv_push_promise<B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        frame: frame::PushPromise,
+    ) -> Result<(), Error> {
+        let id = frame.stream_id();
+        let promised_id = frame.promised_id();
+
+        // First, ensure that the initiating stream is still in a valid state.
+        let parent_key = match self.store.find_mut(&id) {
+            Some(stream) => {
+                // The GOAWAY process has begun. All streams with a greater ID
+                // than specified as part of GOAWAY should be ignored.
+                if id > self.actions.recv.max_stream_id() {
+                    tracing::trace!(
+                        "id ({:?}) > max_stream_id ({:?}), ignoring PUSH_PROMISE",
+                        id,
+                        self.actions.recv.max_stream_id()
+                    );
+                    return Ok(());
+                }
+
+                // The stream must be receive open
+                stream.state.ensure_recv_open()?;
+                stream.key()
+            }
+            None => {
+                proto_err!(conn: "recv_push_promise: initiating stream is in an invalid state");
+                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR).into());
+            }
+        };
+
+        // TODO: Streams in the reserved states do not count towards the concurrency
+        // limit. However, it seems like there should be a cap otherwise this
+        // could grow in memory indefinitely.
+
+        // Ensure that we can reserve streams
+        self.actions.recv.ensure_can_reserve()?;
+
+        // Next, open the stream.
+        //
+        // If `None` is returned, then the stream is being refused. There is no
+        // further work to be done.
+        if self
+            .actions
+            .recv
+            .open(promised_id, Open::PushPromise, &mut self.counts)?
+            .is_none()
+        {
+            return Ok(());
+        }
+
+        // Try to handle the frame and create a corresponding key for the pushed stream
+        // this requires a bit of indirection to make the borrow checker happy.
+        let child_key: Option<store::Key> = {
+            // Create state for the stream
+            let stream = self.store.insert(promised_id, {
+                Stream::new(
+                    promised_id,
+                    self.actions.send.init_window_sz(),
+                    self.actions.recv.init_window_sz(),
+                )
+            });
+
+            let actions = &mut self.actions;
+
+            self.counts.transition(stream, |counts, stream| {
+                let stream_valid = actions.recv.recv_push_promise(frame, stream);
+
+                match stream_valid {
+                    Ok(()) => Ok(Some(stream.key())),
+                    _ => {
+                        let mut send_buffer = send_buffer.inner.lock().unwrap();
+                        actions
+                            .reset_on_recv_stream_err(
+                                &mut *send_buffer,
+                                stream,
+                                counts,
+                                stream_valid,
+                            )
+                            .map(|()| None)
+                    }
+                }
+            })?
+        };
+        // If we're successful, push the headers and stream...
+        if let Some(child) = child_key {
+            let mut ppp = self.store[parent_key].pending_push_promises.take();
+            ppp.push(&mut self.store.resolve(child));
+
+            let parent = &mut self.store.resolve(parent_key);
+            parent.pending_push_promises = ppp;
+            parent.notify_recv();
+        };
+
+        Ok(())
+    }
+
+    fn recv_eof<B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        clear_pending_accept: bool,
+    ) -> Result<(), ()> {
+        let actions = &mut self.actions;
+        let counts = &mut self.counts;
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        if actions.conn_error.is_none() {
+            actions.conn_error = Some(io::Error::from(io::ErrorKind::BrokenPipe).into());
+        }
+
+        tracing::trace!("Streams::recv_eof");
+
+        self.store.for_each(|stream| {
+            counts.transition(stream, |counts, stream| {
+                actions.recv.recv_eof(stream);
+
+                // This handles resetting send state associated with the
+                // stream
+                actions.send.handle_error(send_buffer, stream, counts);
+            })
+        });
+
+        actions.clear_queues(clear_pending_accept, &mut self.store, counts);
+        Ok(())
+    }
+
+    fn poll_complete<T, B>(
+        &mut self,
+        send_buffer: &SendBuffer<B>,
+        cx: &mut Context,
+        dst: &mut Codec<T, Prioritized<B>>,
+    ) -> Poll<io::Result<()>>
+    where
+        T: AsyncWrite + Unpin,
+        B: Buf,
+    {
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        // Send WINDOW_UPDATE frames first
+        //
+        // TODO: It would probably be better to interleave updates w/ data
+        // frames.
+        ready!(self
+            .actions
+            .recv
+            .poll_complete(cx, &mut self.store, &mut self.counts, dst))?;
+
+        // Send any other pending frames
+        ready!(self.actions.send.poll_complete(
+            cx,
+            send_buffer,
+            &mut self.store,
+            &mut self.counts,
+            dst
+        ))?;
+
+        // Nothing else to do, track the task
+        self.actions.task = Some(cx.waker().clone());
+
+        Poll::Ready(Ok(()))
+    }
+
+    fn send_reset<B>(&mut self, send_buffer: &SendBuffer<B>, id: StreamId, reason: Reason) {
+        let key = match self.store.find_entry(id) {
+            Entry::Occupied(e) => e.key(),
+            Entry::Vacant(e) => {
+                // Resetting a stream we don't know about? That could be OK...
+                //
+                // 1. As a server, we just received a request, but that request
+                //    was bad, so we're resetting before even accepting it.
+                //    This is totally fine.
+                //
+                // 2. The remote may have sent us a frame on new stream that
+                //    it's *not* supposed to have done, and thus, we don't know
+                //    the stream. In that case, sending a reset will "open" the
+                //    stream in our store. Maybe that should be a connection
+                //    error instead? At least for now, we need to update what
+                //    our vision of the next stream is.
+                if self.counts.peer().is_local_init(id) {
+                    // We normally would open this stream, so update our
+                    // next-send-id record.
+                    self.actions.send.maybe_reset_next_stream_id(id);
+                } else {
+                    // We normally would recv this stream, so update our
+                    // next-recv-id record.
+                    self.actions.recv.maybe_reset_next_stream_id(id);
+                }
+
+                let stream = Stream::new(id, 0, 0);
+
+                e.insert(stream)
+            }
+        };
+
+        let stream = self.store.resolve(key);
+        let mut send_buffer = send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+        self.actions.send_reset(
+            stream,
+            reason,
+            Initiator::Library,
+            &mut self.counts,
+            send_buffer,
+        );
+    }
+}
+
+impl<B> Streams<B, client::Peer>
+where
+    B: Buf,
+{
+    pub fn poll_pending_open(
+        &mut self,
+        cx: &Context,
+        pending: Option<&OpaqueStreamRef>,
+    ) -> Poll<Result<(), crate::Error>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        me.actions.ensure_no_conn_error()?;
+        me.actions.send.ensure_next_stream_id()?;
+
+        if let Some(pending) = pending {
+            let mut stream = me.store.resolve(pending.key);
+            tracing::trace!("poll_pending_open; stream = {:?}", stream.is_pending_open);
+            if stream.is_pending_open {
+                stream.wait_send(cx);
+                return Poll::Pending;
+            }
+        }
+        Poll::Ready(Ok(()))
+    }
+}
+
+impl<B, P> Streams<B, P>
+where
+    P: Peer,
+{
+    pub fn as_dyn(&self) -> DynStreams<B> {
+        let Self {
+            inner,
+            send_buffer,
+            _p,
+        } = self;
+        DynStreams {
+            inner,
+            send_buffer,
+            peer: P::r#dyn(),
+        }
+    }
+
+    /// This function is safe to call multiple times.
+    ///
+    /// A `Result` is returned to avoid panicking if the mutex is poisoned.
+    pub fn recv_eof(&mut self, clear_pending_accept: bool) -> Result<(), ()> {
+        self.as_dyn().recv_eof(clear_pending_accept)
+    }
+
+    pub(crate) fn max_send_streams(&self) -> usize {
+        self.inner.lock().unwrap().counts.max_send_streams()
+    }
+
+    pub(crate) fn max_recv_streams(&self) -> usize {
+        self.inner.lock().unwrap().counts.max_recv_streams()
+    }
+
+    #[cfg(feature = "unstable")]
+    pub fn num_active_streams(&self) -> usize {
+        let me = self.inner.lock().unwrap();
+        me.store.num_active_streams()
+    }
+
+    pub fn has_streams(&self) -> bool {
+        let me = self.inner.lock().unwrap();
+        me.counts.has_streams()
+    }
+
+    pub fn has_streams_or_other_references(&self) -> bool {
+        let me = self.inner.lock().unwrap();
+        me.counts.has_streams() || me.refs > 1
+    }
+
+    #[cfg(feature = "unstable")]
+    pub fn num_wired_streams(&self) -> usize {
+        let me = self.inner.lock().unwrap();
+        me.store.num_wired_streams()
+    }
+}
+
+// no derive because we don't need B and P to be Clone.
+impl<B, P> Clone for Streams<B, P>
+where
+    P: Peer,
+{
+    fn clone(&self) -> Self {
+        self.inner.lock().unwrap().refs += 1;
+        Streams {
+            inner: self.inner.clone(),
+            send_buffer: self.send_buffer.clone(),
+            _p: ::std::marker::PhantomData,
+        }
+    }
+}
+
+impl<B, P> Drop for Streams<B, P>
+where
+    P: Peer,
+{
+    fn drop(&mut self) {
+        if let Ok(mut inner) = self.inner.lock() {
+            inner.refs -= 1;
+            if inner.refs == 1 {
+                if let Some(task) = inner.actions.task.take() {
+                    task.wake();
+                }
+            }
+        }
+    }
+}
+
+// ===== impl StreamRef =====
+
+impl<B> StreamRef<B> {
+    pub fn send_data(&mut self, data: B, end_stream: bool) -> Result<(), UserError>
+    where
+        B: Buf,
+    {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.opaque.key);
+        let actions = &mut me.actions;
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.counts.transition(stream, |counts, stream| {
+            // Create the data frame
+            let mut frame = frame::Data::new(stream.id, data);
+            frame.set_end_stream(end_stream);
+
+            // Send the data frame
+            actions
+                .send
+                .send_data(frame, send_buffer, stream, counts, &mut actions.task)
+        })
+    }
+
+    pub fn send_trailers(&mut self, trailers: HeaderMap) -> Result<(), UserError> {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.opaque.key);
+        let actions = &mut me.actions;
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.counts.transition(stream, |counts, stream| {
+            // Create the trailers frame
+            let frame = frame::Headers::trailers(stream.id, trailers);
+
+            // Send the trailers frame
+            actions
+                .send
+                .send_trailers(frame, send_buffer, stream, counts, &mut actions.task)
+        })
+    }
+
+    pub fn send_reset(&mut self, reason: Reason) {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.opaque.key);
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.actions
+            .send_reset(stream, reason, Initiator::User, &mut me.counts, send_buffer);
+    }
+
+    pub fn send_response(
+        &mut self,
+        mut response: Response<()>,
+        end_of_stream: bool,
+    ) -> Result<(), UserError> {
+        // Clear before taking lock, incase extensions contain a StreamRef.
+        response.extensions_mut().clear();
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.opaque.key);
+        let actions = &mut me.actions;
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        me.counts.transition(stream, |counts, stream| {
+            let frame = server::Peer::convert_send_message(stream.id, response, end_of_stream);
+
+            actions
+                .send
+                .send_headers(frame, send_buffer, stream, counts, &mut actions.task)
+        })
+    }
+
+    pub fn send_push_promise(
+        &mut self,
+        mut request: Request<()>,
+    ) -> Result<StreamRef<B>, UserError> {
+        // Clear before taking lock, incase extensions contain a StreamRef.
+        request.extensions_mut().clear();
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
+        let send_buffer = &mut *send_buffer;
+
+        let actions = &mut me.actions;
+        let promised_id = actions.send.reserve_local()?;
+
+        let child_key = {
+            let mut child_stream = me.store.insert(
+                promised_id,
+                Stream::new(
+                    promised_id,
+                    actions.send.init_window_sz(),
+                    actions.recv.init_window_sz(),
+                ),
+            );
+            child_stream.state.reserve_local()?;
+            child_stream.is_pending_push = true;
+            child_stream.key()
+        };
+
+        let pushed = {
+            let mut stream = me.store.resolve(self.opaque.key);
+
+            let frame = crate::server::Peer::convert_push_message(stream.id, promised_id, request)?;
+
+            actions
+                .send
+                .send_push_promise(frame, send_buffer, &mut stream, &mut actions.task)
+        };
+
+        if let Err(err) = pushed {
+            let mut child_stream = me.store.resolve(child_key);
+            child_stream.unlink();
+            child_stream.remove();
+            return Err(err.into());
+        }
+
+        me.refs += 1;
+        let opaque =
+            OpaqueStreamRef::new(self.opaque.inner.clone(), &mut me.store.resolve(child_key));
+
+        Ok(StreamRef {
+            opaque,
+            send_buffer: self.send_buffer.clone(),
+        })
+    }
+
+    /// Called by the server after the stream is accepted. Given that clients
+    /// initialize streams by sending HEADERS, the request will always be
+    /// available.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if the request isn't present.
+    pub fn take_request(&self) -> Request<()> {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.opaque.key);
+        me.actions.recv.take_request(&mut stream)
+    }
+
+    /// Called by a client to see if the current stream is pending open
+    pub fn is_pending_open(&self) -> bool {
+        let mut me = self.opaque.inner.lock().unwrap();
+        me.store.resolve(self.opaque.key).is_pending_open
+    }
+
+    /// Request capacity to send data
+    pub fn reserve_capacity(&mut self, capacity: WindowSize) {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.opaque.key);
+
+        me.actions
+            .send
+            .reserve_capacity(capacity, &mut stream, &mut me.counts)
+    }
+
+    /// Returns the stream's current send capacity.
+    pub fn capacity(&self) -> WindowSize {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.opaque.key);
+
+        me.actions.send.capacity(&mut stream)
+    }
+
+    /// Request to be notified when the stream's capacity increases
+    pub fn poll_capacity(&mut self, cx: &Context) -> Poll<Option<Result<WindowSize, UserError>>> {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.opaque.key);
+
+        me.actions.send.poll_capacity(cx, &mut stream)
+    }
+
+    /// Request to be notified for if a `RST_STREAM` is received for this stream.
+    pub(crate) fn poll_reset(
+        &mut self,
+        cx: &Context,
+        mode: proto::PollReset,
+    ) -> Poll<Result<Reason, crate::Error>> {
+        let mut me = self.opaque.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.opaque.key);
+
+        me.actions
+            .send
+            .poll_reset(cx, &mut stream, mode)
+            .map_err(From::from)
+    }
+
+    pub fn clone_to_opaque(&self) -> OpaqueStreamRef
+    where
+        B: 'static,
+    {
+        self.opaque.clone()
+    }
+
+    pub fn stream_id(&self) -> StreamId {
+        self.opaque.stream_id()
+    }
+}
+
+impl<B> Clone for StreamRef<B> {
+    fn clone(&self) -> Self {
+        StreamRef {
+            opaque: self.opaque.clone(),
+            send_buffer: self.send_buffer.clone(),
+        }
+    }
+}
+
+// ===== impl OpaqueStreamRef =====
+
+impl OpaqueStreamRef {
+    fn new(inner: Arc<Mutex<Inner>>, stream: &mut store::Ptr) -> OpaqueStreamRef {
+        stream.ref_inc();
+        OpaqueStreamRef {
+            inner,
+            key: stream.key(),
+        }
+    }
+    /// Called by a client to check for a received response.
+    pub fn poll_response(&mut self, cx: &Context) -> Poll<Result<Response<()>, proto::Error>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions.recv.poll_response(cx, &mut stream)
+    }
+    /// Called by a client to check for a pushed request.
+    pub fn poll_pushed(
+        &mut self,
+        cx: &Context,
+    ) -> Poll<Option<Result<(Request<()>, OpaqueStreamRef), proto::Error>>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+        me.actions
+            .recv
+            .poll_pushed(cx, &mut stream)
+            .map_ok(|(h, key)| {
+                me.refs += 1;
+                let opaque_ref =
+                    OpaqueStreamRef::new(self.inner.clone(), &mut me.store.resolve(key));
+                (h, opaque_ref)
+            })
+    }
+
+    pub fn is_end_stream(&self) -> bool {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let stream = me.store.resolve(self.key);
+
+        me.actions.recv.is_end_stream(&stream)
+    }
+
+    pub fn poll_data(&mut self, cx: &Context) -> Poll<Option<Result<Bytes, proto::Error>>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions.recv.poll_data(cx, &mut stream)
+    }
+
+    pub fn poll_trailers(&mut self, cx: &Context) -> Poll<Option<Result<HeaderMap, proto::Error>>> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions.recv.poll_trailers(cx, &mut stream)
+    }
+
+    pub(crate) fn available_recv_capacity(&self) -> isize {
+        let me = self.inner.lock().unwrap();
+        let me = &*me;
+
+        let stream = &me.store[self.key];
+        stream.recv_flow.available().into()
+    }
+
+    pub(crate) fn used_recv_capacity(&self) -> WindowSize {
+        let me = self.inner.lock().unwrap();
+        let me = &*me;
+
+        let stream = &me.store[self.key];
+        stream.in_flight_recv_data
+    }
+
+    /// Releases recv capacity back to the peer. This may result in sending
+    /// WINDOW_UPDATE frames on both the stream and connection.
+    pub fn release_capacity(&mut self, capacity: WindowSize) -> Result<(), UserError> {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions
+            .recv
+            .release_capacity(capacity, &mut stream, &mut me.actions.task)
+    }
+
+    pub(crate) fn clear_recv_buffer(&mut self) {
+        let mut me = self.inner.lock().unwrap();
+        let me = &mut *me;
+
+        let mut stream = me.store.resolve(self.key);
+
+        me.actions.recv.clear_recv_buffer(&mut stream);
+    }
+
+    pub fn stream_id(&self) -> StreamId {
+        self.inner.lock().unwrap().store[self.key].id
+    }
+}
+
+impl fmt::Debug for OpaqueStreamRef {
+    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+        use std::sync::TryLockError::*;
+
+        match self.inner.try_lock() {
+            Ok(me) => {
+                let stream = &me.store[self.key];
+                fmt.debug_struct("OpaqueStreamRef")
+                    .field("stream_id", &stream.id)
+                    .field("ref_count", &stream.ref_count)
+                    .finish()
+            }
+            Err(Poisoned(_)) => fmt
+                .debug_struct("OpaqueStreamRef")
+                .field("inner", &"<Poisoned>")
+                .finish(),
+            Err(WouldBlock) => fmt
+                .debug_struct("OpaqueStreamRef")
+                .field("inner", &"<Locked>")
+                .finish(),
+        }
+    }
+}
+
+impl Clone for OpaqueStreamRef {
+    fn clone(&self) -> Self {
+        // Increment the ref count
+        let mut inner = self.inner.lock().unwrap();
+        inner.store.resolve(self.key).ref_inc();
+        inner.refs += 1;
+
+        OpaqueStreamRef {
+            inner: self.inner.clone(),
+            key: self.key.clone(),
+        }
+    }
+}
+
+impl Drop for OpaqueStreamRef {
+    fn drop(&mut self) {
+        drop_stream_ref(&self.inner, self.key);
+    }
+}
+
+// TODO: Move back in fn above
+fn drop_stream_ref(inner: &Mutex<Inner>, key: store::Key) {
+    let mut me = match inner.lock() {
+        Ok(inner) => inner,
+        Err(_) => {
+            if ::std::thread::panicking() {
+                tracing::trace!("StreamRef::drop; mutex poisoned");
+                return;
+            } else {
+                panic!("StreamRef::drop; mutex poisoned");
+            }
+        }
+    };
+
+    let me = &mut *me;
+    me.refs -= 1;
+    let mut stream = me.store.resolve(key);
+
+    tracing::trace!("drop_stream_ref; stream={:?}", stream);
+
+    // decrement the stream's ref count by 1.
+    stream.ref_dec();
+
+    let actions = &mut me.actions;
+
+    // If the stream is not referenced and it is already
+    // closed (does not have to go through logic below
+    // of canceling the stream), we should notify the task
+    // (connection) so that it can close properly
+    if stream.ref_count == 0 && stream.is_closed() {
+        if let Some(task) = actions.task.take() {
+            task.wake();
+        }
+    }
+
+    me.counts.transition(stream, |counts, stream| {
+        maybe_cancel(stream, actions, counts);
+
+        if stream.ref_count == 0 {
+            // Release any recv window back to connection, no one can access
+            // it anymore.
+            actions
+                .recv
+                .release_closed_capacity(stream, &mut actions.task);
+
+            // We won't be able to reach our push promises anymore
+            let mut ppp = stream.pending_push_promises.take();
+            while let Some(promise) = ppp.pop(stream.store_mut()) {
+                counts.transition(promise, |counts, stream| {
+                    maybe_cancel(stream, actions, counts);
+                });
+            }
+        }
+    });
+}
+
+fn maybe_cancel(stream: &mut store::Ptr, actions: &mut Actions, counts: &mut Counts) {
+    if stream.is_canceled_interest() {
+        // Server is allowed to early respond without fully consuming the client input stream
+        // But per the RFC, must send a RST_STREAM(NO_ERROR) in such cases. https://www.rfc-editor.org/rfc/rfc7540#section-8.1
+        // Some other http2 implementation may interpret other error code as fatal if not respected (i.e: nginx https://trac.nginx.org/nginx/ticket/2376)
+        let reason = if counts.peer().is_server()
+            && stream.state.is_send_closed()
+            && stream.state.is_recv_streaming()
+        {
+            Reason::NO_ERROR
+        } else {
+            Reason::CANCEL
+        };
+
+        actions
+            .send
+            .schedule_implicit_reset(stream, reason, counts, &mut actions.task);
+        actions.recv.enqueue_reset_expiration(stream, counts);
+    }
+}
+
+// ===== impl SendBuffer =====
+
+impl<B> SendBuffer<B> {
+    fn new() -> Self {
+        let inner = Mutex::new(Buffer::new());
+        SendBuffer { inner }
+    }
+}
+
+// ===== impl Actions =====
+
+impl Actions {
+    fn send_reset<B>(
+        &mut self,
+        stream: store::Ptr,
+        reason: Reason,
+        initiator: Initiator,
+        counts: &mut Counts,
+        send_buffer: &mut Buffer<Frame<B>>,
+    ) {
+        counts.transition(stream, |counts, stream| {
+            self.send.send_reset(
+                reason,
+                initiator,
+                send_buffer,
+                stream,
+                counts,
+                &mut self.task,
+            );
+            self.recv.enqueue_reset_expiration(stream, counts);
+            // if a RecvStream is parked, ensure it's notified
+            stream.notify_recv();
+        });
+    }
+
+    fn reset_on_recv_stream_err<B>(
+        &mut self,
+        buffer: &mut Buffer<Frame<B>>,
+        stream: &mut store::Ptr,
+        counts: &mut Counts,
+        res: Result<(), Error>,
+    ) -> Result<(), Error> {
+        if let Err(Error::Reset(stream_id, reason, initiator)) = res {
+            debug_assert_eq!(stream_id, stream.id);
+            // Reset the stream.
+            self.send
+                .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
+            Ok(())
+        } else {
+            res
+        }
+    }
+
+    fn ensure_not_idle(&mut self, peer: peer::Dyn, id: StreamId) -> Result<(), Reason> {
+        if peer.is_local_init(id) {
+            self.send.ensure_not_idle(id)
+        } else {
+            self.recv.ensure_not_idle(id)
+        }
+    }
+
+    fn ensure_no_conn_error(&self) -> Result<(), proto::Error> {
+        if let Some(ref err) = self.conn_error {
+            Err(err.clone())
+        } else {
+            Ok(())
+        }
+    }
+
+    /// 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(&self, peer: peer::Dyn, id: StreamId) -> bool {
+        if id.is_zero() {
+            return false;
+        }
+        if peer.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, counts: &mut Counts) {
+        self.recv.clear_queues(clear_pending_accept, store, counts);
+        self.send.clear_queues(store, counts);
+    }
+}