forked/h2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor errors internals (#556)

Browse Source 
h2::Error now knows whether protocol errors happened because the user
sent them, because it was received from the remote peer, or because
the library itself emitted an error because it detected a protocol
violation.

It also keeps track of whether it came from a RST_STREAM or GO_AWAY
frame, and in the case of the latter, it includes the additional
debug data if any.

Fixes #530
This commit is contained in:
Anthony Ramine
2021-09-28 18:04:35 +02:00
committed by GitHub
parent cab307d2ed
commit 465f0337f8
26 changed files with 464 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
src/proto/connection.rs
View File

@@ -1,6 +1,6 @@
use crate::codec::{RecvError, UserError};
use crate::codec::UserError;
use crate::frame::{Reason, StreamId};
use crate::{client, frame, proto, server};
use crate::{client, frame, server};
use crate::frame::DEFAULT_INITIAL_WINDOW_SIZE;
use crate::proto::*;
@@ -40,7 +40,7 @@ where
///
/// This exists separately from State in order to support
/// graceful shutdown.
error: Option<Reason>,
error: Option<frame::GoAway>,
/// Pending GOAWAY frames to write.
go_away: GoAway,
@@ -68,7 +68,7 @@ struct DynConnection<'a, B: Buf = Bytes> {
streams: DynStreams<'a, B>,
error: &'a mut Option<Reason>,
error: &'a mut Option<frame::GoAway>,
ping_pong: &'a mut PingPong,
}
@@ -88,10 +88,10 @@ enum State {
Open,
/// The codec must be flushed
Closing(Reason),
Closing(Reason, Initiator),
/// In a closed state
Closed(Reason),
Closed(Reason, Initiator),
}
impl<T, P, B> Connection<T, P, B>
@@ -161,9 +161,9 @@ where
/// Returns `Ready` when the connection is ready to receive a frame.
///
/// Returns `RecvError` as this may raise errors that are caused by delayed
/// Returns `Error` as this may raise errors that are caused by delayed
/// processing of received frames.
fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), RecvError>> {
fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Error>> {
let _e = self.inner.span.enter();
let span = tracing::trace_span!("poll_ready");
let _e = span.enter();
@@ -191,26 +191,24 @@ where
self.inner.as_dyn().go_away_from_user(e)
}
fn take_error(&mut self, ours: Reason) -> Poll<Result<(), proto::Error>> {
let reason = if let Some(theirs) = self.inner.error.take() {
match (ours, theirs) {
// If either side reported an error, return that
// to the user.
(Reason::NO_ERROR, err) | (err, Reason::NO_ERROR) => err,
// If both sides reported an error, give their
// error back to th user. We assume our error
// was a consequence of their error, and less
// important.
(_, theirs) => theirs,
}
} else {
ours
};
fn take_error(&mut self, ours: Reason, initiator: Initiator) -> Result<(), Error> {
let (debug_data, theirs) = self
.inner
.error
.take()
.as_ref()
.map_or((Bytes::new(), Reason::NO_ERROR), |frame| {
(frame.debug_data().clone(), frame.reason())
});
if reason == Reason::NO_ERROR {
Poll::Ready(Ok(()))
} else {
Poll::Ready(Err(proto::Error::Proto(reason)))
match (ours, theirs) {
(Reason::NO_ERROR, Reason::NO_ERROR) => return Ok(()),
(ours, Reason::NO_ERROR) => Err(Error::GoAway(Bytes::new(), ours, initiator)),
// If both sides reported an error, give their
// error back to th user. We assume our error
// was a consequence of their error, and less
// important.
(_, theirs) => Err(Error::remote_go_away(debug_data, theirs)),
}
}
@@ -229,7 +227,7 @@ where
}
/// Advances the internal state of the connection.
pub fn poll(&mut self, cx: &mut Context) -> Poll<Result<(), proto::Error>> {
pub fn poll(&mut self, cx: &mut Context) -> Poll<Result<(), Error>> {
// XXX(eliza): cloning the span is unfortunately necessary here in
// order to placate the borrow checker — `self` is mutably borrowed by
// `poll2`, which means that we can't borrow `self.span` to enter it.
@@ -268,20 +266,22 @@ where
self.inner.as_dyn().handle_poll2_result(result)?
}
State::Closing(reason) => {
State::Closing(reason, initiator) => {
tracing::trace!("connection closing after flush");
// Flush/shutdown the codec
ready!(self.codec.shutdown(cx))?;
// Transition the state to error
self.inner.state = State::Closed(reason);
self.inner.state = State::Closed(reason, initiator);
}
State::Closed(reason, initiator) => {
return Poll::Ready(self.take_error(reason, initiator));
}
State::Closed(reason) => return self.take_error(reason),
}
}
}
fn poll2(&mut self, cx: &mut Context) -> Poll<Result<(), RecvError>> {
fn poll2(&mut self, cx: &mut Context) -> Poll<Result<(), Error>> {
// This happens outside of the loop to prevent needing to do a clock
// check and then comparison of the queue possibly multiple times a
// second (and thus, the clock wouldn't have changed enough to matter).
@@ -300,7 +300,7 @@ where
// the same error back to the user.
return Poll::Ready(Ok(()));
} else {
return Poll::Ready(Err(RecvError::Connection(reason)));
return Poll::Ready(Err(Error::library_go_away(reason)));
}
}
// Only NO_ERROR should be waiting for idle
@@ -384,42 +384,45 @@ where
self.go_away.go_away_from_user(frame);
// Notify all streams of reason we're abruptly closing.
self.streams.recv_err(&proto::Error::Proto(e));
self.streams.handle_error(Error::user_go_away(e));
}
fn handle_poll2_result(&mut self, result: Result<(), RecvError>) -> Result<(), Error> {
use crate::codec::RecvError::*;
fn handle_poll2_result(&mut self, result: Result<(), Error>) -> Result<(), Error> {
match result {
// The connection has shutdown normally
Ok(()) => {
*self.state = State::Closing(Reason::NO_ERROR);
*self.state = State::Closing(Reason::NO_ERROR, Initiator::Library);
Ok(())
}
// Attempting to read a frame resulted in a connection level
// error. This is handled by setting a GOAWAY frame followed by
// terminating the connection.
Err(Connection(e)) => {
Err(Error::GoAway(debug_data, reason, initiator)) => {
let e = Error::GoAway(debug_data, reason, initiator);
tracing::debug!(error = ?e, "Connection::poll; connection error");
// We may have already sent a GOAWAY for this error,
// if so, don't send another, just flush and close up.
if let Some(reason) = self.go_away.going_away_reason() {
if reason == e {
tracing::trace!(" -> already going away");
*self.state = State::Closing(e);
return Ok(());
}
if self
.go_away
.going_away()
.map_or(false, |frame| frame.reason() == reason)
{
tracing::trace!(" -> already going away");
*self.state = State::Closing(reason, initiator);
return Ok(());
}
// Reset all active streams
self.streams.recv_err(&e.into());
self.go_away_now(e);
self.streams.handle_error(e);
self.go_away_now(reason);
Ok(())
}
// Attempting to read a frame resulted in a stream level error.
// This is handled by resetting the frame then trying to read
// another frame.
Err(Stream { id, reason }) => {
Err(Error::Reset(id, reason, initiator)) => {
debug_assert_eq!(initiator, Initiator::Library);
tracing::trace!(?id, ?reason, "stream error");
self.streams.send_reset(id, reason);
Ok(())
@@ -428,12 +431,12 @@ where
// active streams must be reset.
//
// TODO: Are I/O errors recoverable?
Err(Io(e)) => {
Err(Error::Io(e, inner)) => {
tracing::debug!(error = ?e, "Connection::poll; IO error");
let e = e.into();
let e = Error::Io(e, inner);
// Reset all active streams
self.streams.recv_err(&e);
self.streams.handle_error(e.clone());
// Return the error
Err(e)
@@ -441,7 +444,7 @@ where
}
}
fn recv_frame(&mut self, frame: Option<Frame>) -> Result<ReceivedFrame, RecvError> {
fn recv_frame(&mut self, frame: Option<Frame>) -> Result<ReceivedFrame, Error> {
use crate::frame::Frame::*;
match frame {
Some(Headers(frame)) => {
@@ -471,7 +474,7 @@ where
// until they are all EOS. Once they are, State should
// transition to GoAway.
self.streams.recv_go_away(&frame)?;
*self.error = Some(frame.reason());
*self.error = Some(frame);
}
Some(Ping(frame)) => {
tracing::trace!(?frame, "recv PING");

92
src/proto/error.rs
View File

@@ -1,53 +1,87 @@
use crate::codec::{RecvError, SendError};
use crate::frame::Reason;
use crate::codec::SendError;
use crate::frame::{Reason, StreamId};
use bytes::Bytes;
use std::fmt;
use std::io;
/// Either an H2 reason or an I/O error
#[derive(Debug)]
#[derive(Clone, Debug)]
pub enum Error {
Proto(Reason),
Io(io::Error),
Reset(StreamId, Reason, Initiator),
GoAway(Bytes, Reason, Initiator),
Io(io::ErrorKind, Option<String>),
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum Initiator {
User,
Library,
Remote,
}
impl Error {
/// Clone the error for internal purposes.
///
/// `io::Error` is not `Clone`, so we only copy the `ErrorKind`.
pub(super) fn shallow_clone(&self) -> Error {
pub(crate) fn is_local(&self) -> bool {
match *self {
Error::Proto(reason) => Error::Proto(reason),
Error::Io(ref io) => Error::Io(io::Error::from(io.kind())),
Self::Reset(_, _, initiator) | Self::GoAway(_, _, initiator) => initiator.is_local(),
Self::Io(..) => true,
}
}
pub(crate) fn user_go_away(reason: Reason) -> Self {
Self::GoAway(Bytes::new(), reason, Initiator::User)
}
pub(crate) fn library_reset(stream_id: StreamId, reason: Reason) -> Self {
Self::Reset(stream_id, reason, Initiator::Library)
}
pub(crate) fn library_go_away(reason: Reason) -> Self {
Self::GoAway(Bytes::new(), reason, Initiator::Library)
}
pub(crate) fn remote_reset(stream_id: StreamId, reason: Reason) -> Self {
Self::Reset(stream_id, reason, Initiator::Remote)
}
pub(crate) fn remote_go_away(debug_data: Bytes, reason: Reason) -> Self {
Self::GoAway(debug_data, reason, Initiator::Remote)
}
}
impl Initiator {
fn is_local(&self) -> bool {
match *self {
Self::User | Self::Library => true,
Self::Remote => false,
}
}
}
impl From<Reason> for Error {
fn from(src: Reason) -> Self {
Error::Proto(src)
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
match *self {
Self::Reset(_, reason, _) | Self::GoAway(_, reason, _) => reason.fmt(fmt),
Self::Io(_, Some(ref inner)) => inner.fmt(fmt),
Self::Io(kind, None) => io::Error::from(kind).fmt(fmt),
}
}
}
impl From<io::ErrorKind> for Error {
fn from(src: io::ErrorKind) -> Self {
Error::Io(src.into(), None)
}
}
impl From<io::Error> for Error {
fn from(src: io::Error) -> Self {
Error::Io(src)
}
}
impl From<Error> for RecvError {
fn from(src: Error) -> RecvError {
match src {
Error::Proto(reason) => RecvError::Connection(reason),
Error::Io(e) => RecvError::Io(e),
}
Error::Io(src.kind(), src.get_ref().map(|inner| inner.to_string()))
}
}
impl From<Error> for SendError {
fn from(src: Error) -> SendError {
match src {
Error::Proto(reason) => SendError::Connection(reason),
Error::Io(e) => SendError::Io(e),
}
fn from(src: Error) -> Self {
Self::Connection(src)
}
}

16
src/proto/go_away.rs
View File

@@ -31,7 +31,7 @@ pub(super) struct GoAway {
/// well, and we wouldn't want to save that here to accidentally dump in logs,
/// or waste struct space.)
#[derive(Debug)]
struct GoingAway {
pub(crate) struct GoingAway {
/// Stores the highest stream ID of a GOAWAY that has been sent.
///
/// It's illegal to send a subsequent GOAWAY with a higher ID.
@@ -98,9 +98,9 @@ impl GoAway {
self.is_user_initiated
}
/// Return the last Reason we've sent.
pub fn going_away_reason(&self) -> Option<Reason> {
self.going_away.as_ref().map(|g| g.reason)
/// Returns the going away info, if any.
pub fn going_away(&self) -> Option<&GoingAway> {
self.going_away.as_ref()
}
/// Returns if the connection should close now, or wait until idle.
@@ -141,7 +141,7 @@ impl GoAway {
return Poll::Ready(Some(Ok(reason)));
} else if self.should_close_now() {
return match self.going_away_reason() {
return match self.going_away().map(|going_away| going_away.reason) {
Some(reason) => Poll::Ready(Some(Ok(reason))),
None => Poll::Ready(None),
};
@@ -150,3 +150,9 @@ impl GoAway {
Poll::Ready(None)
}
}
impl GoingAway {
pub(crate) fn reason(&self) -> Reason {
self.reason
}
}

2
src/proto/mod.rs
View File

@@ -7,7 +7,7 @@ mod settings;
mod streams;
pub(crate) use self::connection::{Config, Connection};
pub(crate) use self::error::Error;
pub use self::error::{Error, Initiator};
pub(crate) use self::peer::{Dyn as DynPeer, Peer};
pub(crate) use self::ping_pong::UserPings;
pub(crate) use self::streams::{DynStreams, OpaqueStreamRef, StreamRef, Streams};

13
src/proto/peer.rs
View File

@@ -1,7 +1,6 @@
use crate::codec::RecvError;
use crate::error::Reason;
use crate::frame::{Pseudo, StreamId};
use crate::proto::Open;
use crate::proto::{Error, Open};
use http::{HeaderMap, Request, Response};
@@ -21,7 +20,7 @@ pub(crate) trait Peer {
pseudo: Pseudo,
fields: HeaderMap,
stream_id: StreamId,
) -> Result<Self::Poll, RecvError>;
) -> Result<Self::Poll, Error>;
fn is_local_init(id: StreamId) -> bool {
assert!(!id.is_zero());
@@ -61,7 +60,7 @@ impl Dyn {
pseudo: Pseudo,
fields: HeaderMap,
stream_id: StreamId,
) -> Result<PollMessage, RecvError> {
) -> Result<PollMessage, Error> {
if self.is_server() {
crate::server::Peer::convert_poll_message(pseudo, fields, stream_id)
.map(PollMessage::Server)
@@ -72,12 +71,12 @@ impl Dyn {
}
/// Returns true if the remote peer can initiate a stream with the given ID.
pub fn ensure_can_open(&self, id: StreamId, mode: Open) -> Result<(), RecvError> {
pub fn ensure_can_open(&self, id: StreamId, mode: Open) -> Result<(), Error> {
if self.is_server() {
// Ensure that the ID is a valid client initiated ID
if mode.is_push_promise() || !id.is_client_initiated() {
proto_err!(conn: "cannot open stream {:?} - not client initiated", id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
Ok(())
@@ -85,7 +84,7 @@ impl Dyn {
// Ensure that the ID is a valid server initiated ID
if !mode.is_push_promise() || !id.is_server_initiated() {
proto_err!(conn: "cannot open stream {:?} - not server initiated", id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
Ok(())

8
src/proto/settings.rs
View File

@@ -1,4 +1,4 @@
use crate::codec::{RecvError, UserError};
use crate::codec::UserError;
use crate::error::Reason;
use crate::frame;
use crate::proto::*;
@@ -40,7 +40,7 @@ impl Settings {
frame: frame::Settings,
codec: &mut Codec<T, B>,
streams: &mut Streams<C, P>,
) -> Result<(), RecvError>
) -> Result<(), Error>
where
T: AsyncWrite + Unpin,
B: Buf,
@@ -68,7 +68,7 @@ impl Settings {
// We haven't sent any SETTINGS frames to be ACKed, so
// this is very bizarre! Remote is either buggy or malicious.
proto_err!(conn: "received unexpected settings ack");
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
Err(Error::library_go_away(Reason::PROTOCOL_ERROR))
}
}
} else {
@@ -97,7 +97,7 @@ impl Settings {
cx: &mut Context,
dst: &mut Codec<T, B>,
streams: &mut Streams<C, P>,
) -> Poll<Result<(), RecvError>>
) -> Poll<Result<(), Error>>
where
T: AsyncWrite + Unpin,
B: Buf,

5
src/proto/streams/prioritize.rs
View File

@@ -791,7 +791,10 @@ impl Prioritize {
}),
None => {
if let Some(reason) = stream.state.get_scheduled_reset() {
stream.state.set_reset(reason);
let stream_id = stream.id;
stream
.state
.set_reset(stream_id, reason, Initiator::Library);
let frame = frame::Reset::new(stream.id, reason);
Frame::Reset(frame)

99
src/proto/streams/recv.rs
View File

@@ -1,7 +1,7 @@
use super::*;
use crate::codec::{RecvError, UserError};
use crate::frame::{PushPromiseHeaderError, Reason, DEFAULT_INITIAL_WINDOW_SIZE};
use crate::{frame, proto};
use crate::codec::UserError;
use crate::frame::{self, PushPromiseHeaderError, Reason, DEFAULT_INITIAL_WINDOW_SIZE};
use crate::proto::{self, Error};
use std::task::Context;
use http::{HeaderMap, Request, Response};
@@ -68,7 +68,7 @@ pub(super) enum Event {
#[derive(Debug)]
pub(super) enum RecvHeaderBlockError<T> {
Oversize(T),
State(RecvError),
State(Error),
}
#[derive(Debug)]
@@ -124,7 +124,7 @@ impl Recv {
id: StreamId,
mode: Open,
counts: &mut Counts,
) -> Result<Option<StreamId>, RecvError> {
) -> Result<Option<StreamId>, Error> {
assert!(self.refused.is_none());
counts.peer().ensure_can_open(id, mode)?;
@@ -132,7 +132,7 @@ impl Recv {
let next_id = self.next_stream_id()?;
if id < next_id {
proto_err!(conn: "id ({:?}) < next_id ({:?})", id, next_id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
self.next_stream_id = id.next_id();
@@ -176,11 +176,7 @@ impl Recv {
Ok(v) => v,
Err(()) => {
proto_err!(stream: "could not parse content-length; stream={:?}", stream.id);
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
}
.into());
return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR).into());
}
};
@@ -312,16 +308,13 @@ impl Recv {
&mut self,
frame: frame::Headers,
stream: &mut store::Ptr,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
// Transition the state
stream.state.recv_close()?;
if stream.ensure_content_length_zero().is_err() {
proto_err!(stream: "recv_trailers: content-length is not zero; stream={:?};", stream.id);
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
});
return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
}
let trailers = frame.into_fields();
@@ -455,7 +448,7 @@ impl Recv {
&mut self,
settings: &frame::Settings,
store: &mut Store,
) -> Result<(), RecvError> {
) -> Result<(), proto::Error> {
let target = if let Some(val) = settings.initial_window_size() {
val
} else {
@@ -502,7 +495,7 @@ impl Recv {
stream
.recv_flow
.inc_window(inc)
.map_err(RecvError::Connection)?;
.map_err(proto::Error::library_go_away)?;
stream.recv_flow.assign_capacity(inc);
Ok(())
})
@@ -520,11 +513,7 @@ impl Recv {
stream.pending_recv.is_empty()
}
pub fn recv_data(
&mut self,
frame: frame::Data,
stream: &mut store::Ptr,
) -> Result<(), RecvError> {
pub fn recv_data(&mut self, frame: frame::Data, stream: &mut store::Ptr) -> Result<(), Error> {
let sz = frame.payload().len();
// This should have been enforced at the codec::FramedRead layer, so
@@ -542,7 +531,7 @@ impl Recv {
// Receiving a DATA frame when not expecting one is a protocol
// error.
proto_err!(conn: "unexpected DATA frame; stream={:?}", stream.id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
tracing::trace!(
@@ -557,7 +546,7 @@ impl Recv {
"recv_data; frame ignored on locally reset {:?} for some time",
stream.id,
);
return self.ignore_data(sz);
return Ok(self.ignore_data(sz)?);
}
// Ensure that there is enough capacity on the connection before acting
@@ -573,10 +562,7 @@ impl Recv {
// So, for violating the **stream** window, we can send either a
// stream or connection error. We've opted to send a stream
// error.
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::FLOW_CONTROL_ERROR,
});
return Err(Error::library_reset(stream.id, Reason::FLOW_CONTROL_ERROR));
}
if stream.dec_content_length(frame.payload().len()).is_err() {
@@ -585,10 +571,7 @@ impl Recv {
stream.id,
frame.payload().len(),
);
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
});
return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
}
if frame.is_end_stream() {
@@ -598,15 +581,12 @@ impl Recv {
stream.id,
frame.payload().len(),
);
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
});
return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
}
if stream.state.recv_close().is_err() {
proto_err!(conn: "recv_data: failed to transition to closed state; stream={:?}", stream.id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR).into());
}
}
@@ -625,7 +605,7 @@ impl Recv {
Ok(())
}
pub fn ignore_data(&mut self, sz: WindowSize) -> Result<(), RecvError> {
pub fn ignore_data(&mut self, sz: WindowSize) -> Result<(), Error> {
// Ensure that there is enough capacity on the connection...
self.consume_connection_window(sz)?;
@@ -641,14 +621,14 @@ impl Recv {
Ok(())
}
pub fn consume_connection_window(&mut self, sz: WindowSize) -> Result<(), RecvError> {
pub fn consume_connection_window(&mut self, sz: WindowSize) -> Result<(), Error> {
if self.flow.window_size() < sz {
tracing::debug!(
"connection error FLOW_CONTROL_ERROR -- window_size ({:?}) < sz ({:?});",
self.flow.window_size(),
sz,
);
return Err(RecvError::Connection(Reason::FLOW_CONTROL_ERROR));
return Err(Error::library_go_away(Reason::FLOW_CONTROL_ERROR));
}
// Update connection level flow control
@@ -663,7 +643,7 @@ impl Recv {
&mut self,
frame: frame::PushPromise,
stream: &mut store::Ptr,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
stream.state.reserve_remote()?;
if frame.is_over_size() {
// A frame is over size if the decoded header block was bigger than
@@ -682,10 +662,10 @@ impl Recv {
headers frame is over size; promised_id={:?};",
frame.promised_id(),
);
return Err(RecvError::Stream {
id: frame.promised_id(),
reason: Reason::REFUSED_STREAM,
});
return Err(Error::library_reset(
frame.promised_id(),
Reason::REFUSED_STREAM,
));
}
let promised_id = frame.promised_id();
@@ -708,10 +688,7 @@ impl Recv {
promised_id,
),
}
return Err(RecvError::Stream {
id: promised_id,
reason: Reason::PROTOCOL_ERROR,
});
return Err(Error::library_reset(promised_id, Reason::PROTOCOL_ERROR));
}
use super::peer::PollMessage::*;
@@ -741,18 +718,16 @@ impl Recv {
/// Handle remote sending an explicit RST_STREAM.
pub fn recv_reset(&mut self, frame: frame::Reset, stream: &mut Stream) {
// Notify the stream
stream
.state
.recv_reset(frame.reason(), stream.is_pending_send);
stream.state.recv_reset(frame, stream.is_pending_send);
stream.notify_send();
stream.notify_recv();
}
/// Handle a received error
pub fn recv_err(&mut self, err: &proto::Error, stream: &mut Stream) {
/// Handle a connection-level error
pub fn handle_error(&mut self, err: &proto::Error, stream: &mut Stream) {
// Receive an error
stream.state.recv_err(err);
stream.state.handle_error(err);
// If a receiver is waiting, notify it
stream.notify_send();
@@ -783,11 +758,11 @@ impl Recv {
self.max_stream_id
}
pub fn next_stream_id(&self) -> Result<StreamId, RecvError> {
pub fn next_stream_id(&self) -> Result<StreamId, Error> {
if let Ok(id) = self.next_stream_id {
Ok(id)
} else {
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
Err(Error::library_go_away(Reason::PROTOCOL_ERROR))
}
}
@@ -802,10 +777,10 @@ impl Recv {
}
/// Returns true if the remote peer can reserve a stream with the given ID.
pub fn ensure_can_reserve(&self) -> Result<(), RecvError> {
pub fn ensure_can_reserve(&self) -> Result<(), Error> {
if !self.is_push_enabled {
proto_err!(conn: "recv_push_promise: push is disabled");
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
Ok(())
@@ -1092,8 +1067,8 @@ impl Open {
// ===== impl RecvHeaderBlockError =====
impl<T> From<RecvError> for RecvHeaderBlockError<T> {
fn from(err: RecvError) -> Self {
impl<T> From<Error> for RecvHeaderBlockError<T> {
fn from(err: Error) -> Self {
RecvHeaderBlockError::State(err)
}
}

37
src/proto/streams/send.rs
View File

@@ -2,8 +2,9 @@ use super::{
store, Buffer, Codec, Config, Counts, Frame, Prioritize, Prioritized, Store, Stream, StreamId,
StreamIdOverflow, WindowSize,
};
use crate::codec::{RecvError, UserError};
use crate::codec::UserError;
use crate::frame::{self, Reason};
use crate::proto::{Error, Initiator};
use bytes::Buf;
use http;
@@ -161,6 +162,7 @@ impl Send {
pub fn send_reset<B>(
&mut self,
reason: Reason,
initiator: Initiator,
buffer: &mut Buffer<Frame<B>>,
stream: &mut store::Ptr,
counts: &mut Counts,
@@ -169,14 +171,16 @@ impl Send {
let is_reset = stream.state.is_reset();
let is_closed = stream.state.is_closed();
let is_empty = stream.pending_send.is_empty();
let stream_id = stream.id;
tracing::trace!(
"send_reset(..., reason={:?}, stream={:?}, ..., \
"send_reset(..., reason={:?}, initiator={:?}, stream={:?}, ..., \
is_reset={:?}; is_closed={:?}; pending_send.is_empty={:?}; \
state={:?} \
",
reason,
stream.id,
initiator,
stream_id,
is_reset,
is_closed,
is_empty,
@@ -187,13 +191,13 @@ impl Send {
// Don't double reset
tracing::trace!(
" -> not sending RST_STREAM ({:?} is already reset)",
stream.id
stream_id
);
return;
}
// Transition the state to reset no matter what.
stream.state.set_reset(reason);
stream.state.set_reset(stream_id, reason, initiator);
// If closed AND the send queue is flushed, then the stream cannot be
// reset explicitly, either. Implicit resets can still be queued.
@@ -201,7 +205,7 @@ impl Send {
tracing::trace!(
" -> not sending explicit RST_STREAM ({:?} was closed \
and send queue was flushed)",
stream.id
stream_id
);
return;
}
@@ -371,7 +375,14 @@ impl Send {
if let Err(e) = self.prioritize.recv_stream_window_update(sz, stream) {
tracing::debug!("recv_stream_window_update !!; err={:?}", e);
self.send_reset(Reason::FLOW_CONTROL_ERROR, buffer, stream, counts, task);
self.send_reset(
Reason::FLOW_CONTROL_ERROR,
Initiator::Library,
buffer,
stream,
counts,
task,
);
return Err(e);
}
@@ -379,7 +390,7 @@ impl Send {
Ok(())
}
pub(super) fn recv_go_away(&mut self, last_stream_id: StreamId) -> Result<(), RecvError> {
pub(super) fn recv_go_away(&mut self, last_stream_id: StreamId) -> Result<(), Error> {
if last_stream_id > self.max_stream_id {
// The remote endpoint sent a `GOAWAY` frame indicating a stream
// that we never sent, or that we have already terminated on account
@@ -392,14 +403,14 @@ impl Send {
"recv_go_away: last_stream_id ({:?}) > max_stream_id ({:?})",
last_stream_id, self.max_stream_id,
);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
self.max_stream_id = last_stream_id;
Ok(())
}
pub fn recv_err<B>(
pub fn handle_error<B>(
&mut self,
buffer: &mut Buffer<Frame<B>>,
stream: &mut store::Ptr,
@@ -417,7 +428,7 @@ impl Send {
store: &mut Store,
counts: &mut Counts,
task: &mut Option<Waker>,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
// Applies an update to the remote endpoint's initial window size.
//
// Per RFC 7540 §6.9.2:
@@ -480,7 +491,7 @@ impl Send {
// of a stream is reduced? Maybe it should if the capacity
// is reduced to zero, allowing the producer to stop work.
Ok::<_, RecvError>(())
Ok::<_, Error>(())
})?;
self.prioritize
@@ -490,7 +501,7 @@ impl Send {
store.for_each(|mut stream| {
self.recv_stream_window_update(inc, buffer, &mut stream, counts, task)
.map_err(RecvError::Connection)
.map_err(Error::library_go_away)
})?;
}
}

103
src/proto/streams/state.rs
View File

@@ -1,9 +1,8 @@
use std::io;
use crate::codec::UserError::*;
use crate::codec::{RecvError, UserError};
use crate::frame::{self, Reason};
use crate::proto::{self, PollReset};
use crate::codec::UserError;
use crate::frame::{self, Reason, StreamId};
use crate::proto::{self, Error, Initiator, PollReset};
use self::Inner::*;
use self::Peer::*;
@@ -53,7 +52,7 @@ pub struct State {
inner: Inner,
}
#[derive(Debug, Clone, Copy)]
#[derive(Debug, Clone)]
enum Inner {
Idle,
// TODO: these states shouldn't count against concurrency limits:
@@ -71,12 +70,10 @@ enum Peer {
Streaming,
}
#[derive(Debug, Copy, Clone)]
#[derive(Debug, Clone)]
enum Cause {
EndStream,
Proto(Reason),
LocallyReset(Reason),
Io,
Error(Error),
/// This indicates to the connection that a reset frame must be sent out
/// once the send queue has been flushed.
@@ -85,7 +82,7 @@ enum Cause {
/// - User drops all references to a stream, so we want to CANCEL the it.
/// - Header block size was too large, so we want to REFUSE, possibly
/// after sending a 431 response frame.
Scheduled(Reason),
ScheduledLibraryReset(Reason),
}
impl State {
@@ -123,7 +120,7 @@ impl State {
}
_ => {
// All other transitions result in a protocol error
return Err(UnexpectedFrameType);
return Err(UserError::UnexpectedFrameType);
}
};
@@ -133,7 +130,7 @@ impl State {
/// Opens the receive-half of the stream when a HEADERS frame is received.
///
/// Returns true if this transitions the state to Open.
pub fn recv_open(&mut self, frame: &frame::Headers) -> Result<bool, RecvError> {
pub fn recv_open(&mut self, frame: &frame::Headers) -> Result<bool, Error> {
let mut initial = false;
let eos = frame.is_end_stream();
@@ -195,10 +192,10 @@ impl State {
HalfClosedLocal(Streaming)
}
}
state => {
ref state => {
// All other transitions result in a protocol error
proto_err!(conn: "recv_open: in unexpected state {:?}", state);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
};
@@ -206,15 +203,15 @@ impl State {
}
/// Transition from Idle -> ReservedRemote
pub fn reserve_remote(&mut self) -> Result<(), RecvError> {
pub fn reserve_remote(&mut self) -> Result<(), Error> {
match self.inner {
Idle => {
self.inner = ReservedRemote;
Ok(())
}
state => {
ref state => {
proto_err!(conn: "reserve_remote: in unexpected state {:?}", state);
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
Err(Error::library_go_away(Reason::PROTOCOL_ERROR))
}
}
}
@@ -231,7 +228,7 @@ impl State {
}
/// Indicates that the remote side will not send more data to the local.
pub fn recv_close(&mut self) -> Result<(), RecvError> {
pub fn recv_close(&mut self) -> Result<(), Error> {
match self.inner {
Open { local, .. } => {
// The remote side will continue to receive data.
@@ -244,9 +241,9 @@ impl State {
self.inner = Closed(Cause::EndStream);
Ok(())
}
state => {
ref state => {
proto_err!(conn: "recv_close: in unexpected state {:?}", state);
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
Err(Error::library_go_away(Reason::PROTOCOL_ERROR))
}
}
}
@@ -254,9 +251,9 @@ impl State {
/// The remote explicitly sent a RST_STREAM.
///
/// # Arguments
/// - `reason`: the reason field of the received RST_STREAM frame.
/// - `frame`: the received RST_STREAM frame.
/// - `queued`: true if this stream has frames in the pending send queue.
pub fn recv_reset(&mut self, reason: Reason, queued: bool) {
pub fn recv_reset(&mut self, frame: frame::Reset, queued: bool) {
match self.inner {
// If the stream is already in a `Closed` state, do nothing,
// provided that there are no frames still in the send queue.
@@ -275,30 +272,28 @@ impl State {
// In either of these cases, we want to overwrite the stream's
// previous state with the received RST_STREAM, so that the queue
// will be cleared by `Prioritize::pop_frame`.
state => {
ref state => {
tracing::trace!(
"recv_reset; reason={:?}; state={:?}; queued={:?}",
reason,
"recv_reset; frame={:?}; state={:?}; queued={:?}",
frame,
state,
queued
);
self.inner = Closed(Cause::Proto(reason));
self.inner = Closed(Cause::Error(Error::remote_reset(
frame.stream_id(),
frame.reason(),
)));
}
}
}
/// We noticed a protocol error.
pub fn recv_err(&mut self, err: &proto::Error) {
use crate::proto::Error::*;
/// Handle a connection-level error.
pub fn handle_error(&mut self, err: &proto::Error) {
match self.inner {
Closed(..) => {}
_ => {
tracing::trace!("recv_err; err={:?}", err);
self.inner = Closed(match *err {
Proto(reason) => Cause::LocallyReset(reason),
Io(..) => Cause::Io,
});
tracing::trace!("handle_error; err={:?}", err);
self.inner = Closed(Cause::Error(err.clone()));
}
}
}
@@ -306,9 +301,9 @@ impl State {
pub fn recv_eof(&mut self) {
match self.inner {
Closed(..) => {}
s => {
tracing::trace!("recv_eof; state={:?}", s);
self.inner = Closed(Cause::Io);
ref state => {
tracing::trace!("recv_eof; state={:?}", state);
self.inner = Closed(Cause::Error(io::ErrorKind::BrokenPipe.into()));
}
}
}
@@ -325,39 +320,39 @@ impl State {
tracing::trace!("send_close: HalfClosedRemote => Closed");
self.inner = Closed(Cause::EndStream);
}
state => panic!("send_close: unexpected state {:?}", state),
ref state => panic!("send_close: unexpected state {:?}", state),
}
}
/// Set the stream state to reset locally.
pub fn set_reset(&mut self, reason: Reason) {
self.inner = Closed(Cause::LocallyReset(reason));
pub fn set_reset(&mut self, stream_id: StreamId, reason: Reason, initiator: Initiator) {
self.inner = Closed(Cause::Error(Error::Reset(stream_id, reason, initiator)));
}
/// Set the stream state to a scheduled reset.
pub fn set_scheduled_reset(&mut self, reason: Reason) {
debug_assert!(!self.is_closed());
self.inner = Closed(Cause::Scheduled(reason));
self.inner = Closed(Cause::ScheduledLibraryReset(reason));
}
pub fn get_scheduled_reset(&self) -> Option<Reason> {
match self.inner {
Closed(Cause::Scheduled(reason)) => Some(reason),
Closed(Cause::ScheduledLibraryReset(reason)) => Some(reason),
_ => None,
}
}
pub fn is_scheduled_reset(&self) -> bool {
match self.inner {
Closed(Cause::Scheduled(..)) => true,
Closed(Cause::ScheduledLibraryReset(..)) => true,
_ => false,
}
}
pub fn is_local_reset(&self) -> bool {
match self.inner {
Closed(Cause::LocallyReset(_)) => true,
Closed(Cause::Scheduled(..)) => true,
Closed(Cause::Error(ref e)) => e.is_local(),
Closed(Cause::ScheduledLibraryReset(..)) => true,
_ => false,
}
}
@@ -436,10 +431,10 @@ impl State {
pub fn ensure_recv_open(&self) -> Result<bool, proto::Error> {
// TODO: Is this correct?
match self.inner {
Closed(Cause::Proto(reason))
| Closed(Cause::LocallyReset(reason))
| Closed(Cause::Scheduled(reason)) => Err(proto::Error::Proto(reason)),
Closed(Cause::Io) => Err(proto::Error::Io(io::ErrorKind::BrokenPipe.into())),
Closed(Cause::Error(ref e)) => Err(e.clone()),
Closed(Cause::ScheduledLibraryReset(reason)) => {
Err(proto::Error::library_go_away(reason))
}
Closed(Cause::EndStream) | HalfClosedRemote(..) | ReservedLocal => Ok(false),
_ => Ok(true),
}
@@ -448,10 +443,10 @@ impl State {
/// Returns a reason if the stream has been reset.
pub(super) fn ensure_reason(&self, mode: PollReset) -> Result<Option<Reason>, crate::Error> {
match self.inner {
Closed(Cause::Proto(reason))
| Closed(Cause::LocallyReset(reason))
| Closed(Cause::Scheduled(reason)) => Ok(Some(reason)),
Closed(Cause::Io) => Err(proto::Error::Io(io::ErrorKind::BrokenPipe.into()).into()),
Closed(Cause::Error(Error::Reset(_, reason, _)))
| Closed(Cause::Error(Error::GoAway(_, reason, _)))
| Closed(Cause::ScheduledLibraryReset(reason)) => Ok(Some(reason)),
Closed(Cause::Error(ref e)) => Err(e.clone().into()),
Open {
local: Streaming, ..
}

123
src/proto/streams/streams.rs
View File

@@ -1,9 +1,9 @@
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, RecvError, SendError, UserError};
use crate::codec::{Codec, SendError, UserError};
use crate::frame::{self, Frame, Reason};
use crate::proto::{peer, Open, Peer, WindowSize};
use crate::proto::{peer, Error, Initiator, Open, Peer, WindowSize};
use crate::{client, proto, server};
use bytes::{Buf, Bytes};
@@ -180,7 +180,7 @@ where
me.poll_complete(&self.send_buffer, cx, dst)
}
pub fn apply_remote_settings(&mut self, frame: &frame::Settings) -> Result<(), RecvError> {
pub fn apply_remote_settings(&mut self, frame: &frame::Settings) -> Result<(), Error> {
let mut me = self.inner.lock().unwrap();
let me = &mut *me;
@@ -198,7 +198,7 @@ where
)
}
pub fn apply_local_settings(&mut self, frame: &frame::Settings) -> Result<(), RecvError> {
pub fn apply_local_settings(&mut self, frame: &frame::Settings) -> Result<(), Error> {
let mut me = self.inner.lock().unwrap();
let me = &mut *me;
@@ -297,30 +297,30 @@ where
}
impl<B> DynStreams<'_, B> {
pub fn recv_headers(&mut self, frame: frame::Headers) -> Result<(), RecvError> {
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<(), RecvError> {
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<(), RecvError> {
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)
}
/// Handle a received error and return the ID of the last processed stream.
pub fn recv_err(&mut self, err: &proto::Error) -> StreamId {
/// 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.recv_err(&self.send_buffer, err)
me.handle_error(&self.send_buffer, err)
}
pub fn recv_go_away(&mut self, frame: &frame::GoAway) -> Result<(), RecvError> {
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)
}
@@ -329,12 +329,12 @@ impl<B> DynStreams<'_, B> {
self.inner.lock().unwrap().actions.recv.last_processed_id()
}
pub fn recv_window_update(&mut self, frame: frame::WindowUpdate) -> Result<(), RecvError> {
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<(), RecvError> {
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)
}
@@ -375,7 +375,7 @@ impl Inner {
peer: peer::Dyn,
send_buffer: &SendBuffer<B>,
frame: frame::Headers,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let id = frame.stream_id();
// The GOAWAY process has begun. All streams with a greater ID than
@@ -405,10 +405,7 @@ impl Inner {
"recv_headers for old stream={:?}, sending STREAM_CLOSED",
id,
);
return Err(RecvError::Stream {
id,
reason: Reason::STREAM_CLOSED,
});
return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
}
}
@@ -471,10 +468,7 @@ impl Inner {
Ok(())
} else {
Err(RecvError::Stream {
id: stream.id,
reason: Reason::REFUSED_STREAM,
})
Err(Error::library_reset(stream.id, Reason::REFUSED_STREAM))
}
},
Err(RecvHeaderBlockError::State(err)) => Err(err),
@@ -484,10 +478,7 @@ impl Inner {
// 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(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
});
return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
}
actions.recv.recv_trailers(frame, stream)
@@ -502,7 +493,7 @@ impl Inner {
peer: peer::Dyn,
send_buffer: &SendBuffer<B>,
frame: frame::Data,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let id = frame.stream_id();
let stream = match self.store.find_mut(&id) {
@@ -529,14 +520,11 @@ impl Inner {
let sz = sz as WindowSize;
self.actions.recv.ignore_data(sz)?;
return Err(RecvError::Stream {
id,
reason: Reason::STREAM_CLOSED,
});
return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
}
proto_err!(conn: "recv_data: stream not found; id={:?}", id);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
};
@@ -551,7 +539,7 @@ impl Inner {
// 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(RecvError::Stream { .. }) = res {
if let Err(Error::Reset(..)) = res {
actions
.recv
.release_connection_capacity(sz as WindowSize, &mut None);
@@ -564,12 +552,12 @@ impl Inner {
&mut self,
send_buffer: &SendBuffer<B>,
frame: frame::Reset,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let id = frame.stream_id();
if id.is_zero() {
proto_err!(conn: "recv_reset: invalid stream ID 0");
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
}
// The GOAWAY process has begun. All streams with a greater ID than
@@ -589,7 +577,7 @@ impl Inner {
// TODO: Are there other error cases?
self.actions
.ensure_not_idle(self.counts.peer(), id)
.map_err(RecvError::Connection)?;
.map_err(Error::library_go_away)?;
return Ok(());
}
@@ -602,7 +590,7 @@ impl Inner {
self.counts.transition(stream, |counts, stream| {
actions.recv.recv_reset(frame, stream);
actions.send.recv_err(send_buffer, stream, counts);
actions.send.handle_error(send_buffer, stream, counts);
assert!(stream.state.is_closed());
Ok(())
})
@@ -612,7 +600,7 @@ impl Inner {
&mut self,
send_buffer: &SendBuffer<B>,
frame: frame::WindowUpdate,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let id = frame.stream_id();
let mut send_buffer = send_buffer.inner.lock().unwrap();
@@ -622,7 +610,7 @@ impl Inner {
self.actions
.send
.recv_connection_window_update(frame, &mut self.store, &mut self.counts)
.map_err(RecvError::Connection)?;
.map_err(Error::library_go_away)?;
} else {
// The remote may send window updates for streams that the local now
// considers closed. It's ok...
@@ -640,14 +628,14 @@ impl Inner {
} else {
self.actions
.ensure_not_idle(self.counts.peer(), id)
.map_err(RecvError::Connection)?;
.map_err(Error::library_go_away)?;
}
}
Ok(())
}
fn recv_err<B>(&mut self, send_buffer: &SendBuffer<B>, err: &proto::Error) -> StreamId {
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();
@@ -658,14 +646,14 @@ impl Inner {
self.store
.for_each(|stream| {
counts.transition(stream, |counts, stream| {
actions.recv.recv_err(err, &mut *stream);
actions.send.recv_err(send_buffer, stream, counts);
actions.recv.handle_error(&err, &mut *stream);
actions.send.handle_error(send_buffer, stream, counts);
Ok::<_, ()>(())
})
})
.unwrap();
actions.conn_error = Some(err.shallow_clone());
actions.conn_error = Some(err);
last_processed_id
}
@@ -674,7 +662,7 @@ impl Inner {
&mut self,
send_buffer: &SendBuffer<B>,
frame: &frame::GoAway,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let actions = &mut self.actions;
let counts = &mut self.counts;
let mut send_buffer = send_buffer.inner.lock().unwrap();
@@ -684,14 +672,14 @@ impl Inner {
actions.send.recv_go_away(last_stream_id)?;
let err = frame.reason().into();
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.recv_err(&err, &mut *stream);
actions.send.recv_err(send_buffer, stream, counts);
actions.recv.handle_error(&err, &mut *stream);
actions.send.handle_error(send_buffer, stream, counts);
Ok::<_, ()>(())
})
} else {
@@ -709,7 +697,7 @@ impl Inner {
&mut self,
send_buffer: &SendBuffer<B>,
frame: frame::PushPromise,
) -> Result<(), RecvError> {
) -> Result<(), Error> {
let id = frame.stream_id();
let promised_id = frame.promised_id();
@@ -733,7 +721,7 @@ impl Inner {
}
None => {
proto_err!(conn: "recv_push_promise: initiating stream is in an invalid state");
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
return Err(Error::library_go_away(Reason::PROTOCOL_ERROR).into());
}
};
@@ -826,7 +814,7 @@ impl Inner {
// This handles resetting send state associated with the
// stream
actions.send.recv_err(send_buffer, stream, counts);
actions.send.handle_error(send_buffer, stream, counts);
Ok::<_, ()>(())
})
})
@@ -886,8 +874,13 @@ impl Inner {
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, &mut self.counts, send_buffer);
self.actions.send_reset(
stream,
reason,
Initiator::Library,
&mut self.counts,
send_buffer,
);
}
}
@@ -1060,7 +1053,7 @@ impl<B> StreamRef<B> {
let send_buffer = &mut *send_buffer;
me.actions
.send_reset(stream, reason, &mut me.counts, send_buffer);
.send_reset(stream, reason, Initiator::User, &mut me.counts, send_buffer);
}
pub fn send_response(
@@ -1468,12 +1461,19 @@ impl Actions {
&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, send_buffer, stream, counts, &mut self.task);
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();
@@ -1485,12 +1485,13 @@ impl Actions {
buffer: &mut Buffer<Frame<B>>,
stream: &mut store::Ptr,
counts: &mut Counts,
res: Result<(), RecvError>,
) -> Result<(), RecvError> {
if let Err(RecvError::Stream { reason, .. }) = res {
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, buffer, stream, counts, &mut self.task);
.send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
Ok(())
} else {
res
@@ -1507,7 +1508,7 @@ impl Actions {
fn ensure_no_conn_error(&self) -> Result<(), proto::Error> {
if let Some(ref err) = self.conn_error {
Err(err.shallow_clone())
Err(err.clone())
} else {
Ok(())
}