irpc 1.0.0

//! Module for cross-process RPC.
//!
//! The code in this module works with all connections that use [`noq`]
//! streams. The [`crate::noq`] module has the transport with dial by socket
//! address. The [`crate::iroh`] module has the transport with dial by endpoint id.
//!
//! # Requests and channel messages
//!
//! Each request has its own pair of streams. The client writes the request as
//! the first message on its stream. After the request, the streams carry the
//! channel messages of the request: the response, and for some requests,
//! updates from the client. A `Request` error code is about the request, and a
//! `Message` error code is about a channel message, see [`ErrorCode`].
//!
//! # Bad requests
//!
//! A request is bad if it is larger than [`MAX_MESSAGE_SIZE`], or if it does
//! not decode. For example, a client of a newer protocol version sent a request
//! type that the server does not know. A stream that ends before its request is
//! complete also counts as a request that does not decode. The server stops and
//! resets the streams of a bad request with
//! [`ErrorCode::RequestMaxSizeExceeded`] or [`ErrorCode::RequestDecodeFailed`].
//!
//! By default, a [`Handler`] then closes the connection with the same code, as
//! for any other protocol violation. A protocol that adds request types at the
//! end of its enum can use [`Handler::skip_bad_requests`]: the handler then
//! reads the next request, so an old server rejects the new requests and keeps
//! serving the old ones on the same connection.
//!
//! The client gets the code in an [`RpcRecvError::Remote`] from the receiver of
//! the response, for example from [`Client::rpc`], and in an
//! [`RpcSendError::Remote`] from the sender of its updates. When the server also
//! closes the connection, as a [`Handler`] does by default, the close usually
//! arrives first, and the client gets a `ConnectionClosed` with the code
//! instead. In both cases,
//! `remote_code` returns the code, for example [`crate::Error::remote_code`].
//!
//! [`read_request`] returns a bad request as a [`ReadRequestError`]. A server
//! with its own loop decides what to do: close the connection, or read the
//! next request.
//!
//! # Abandoned requests
//!
//! A client that drops a request before it wrote all of it resets the stream
//! with [`ErrorCode::RequestAborted`]. For example, it dropped the future of
//! [`Client::rpc`] while it wrote a large request. A client that cannot send a
//! request resets the stream with [`ErrorCode::RequestMaxSizeExceeded`] or
//! [`ErrorCode::RequestEncodeFailed`].
//!
//! When the client resets the stream of a request before the server read the
//! request, the server skips the request and reads the next one. It resets its
//! side of the stream with [`ErrorCode::RequestAborted`], so that the receiver
//! of the response gets an error instead of the end of the stream.
//!
//! # Channel messages
//!
//! A sender resets its stream with [`ErrorCode::MessageMaxSizeExceeded`] if a
//! message is larger than [`MAX_MESSAGE_SIZE`], and with
//! [`ErrorCode::MessageEncodeFailed`] if a message does not encode. A receiver
//! stops its stream with [`ErrorCode::MessageMaxSizeExceeded`] if a message is
//! larger than [`MAX_MESSAGE_SIZE`], and with [`ErrorCode::MessageDecodeFailed`]
//! if a message does not decode. The channel then ends. The side that found the
//! error gets its own variant, for example [`RpcRecvError::DecodeFailed`], and the
//! other side gets the code in a `Remote` variant, [`RpcSendError::Remote`] or
//! [`RpcRecvError::Remote`]. A reset can drop messages that the receiver did not
//! read yet.
//!
//! A oneshot sender that drops its message before it wrote all of it, for
//! example because the future of the send was dropped, resets its stream with
//! [`ErrorCode::MessageAborted`], and the receiver gets
//! [`oneshot::RecvError::SenderClosed`], as for a sender that was dropped.
//!
//! An mpsc frame holds the message and a marker after it, which says whether
//! the receiver delivers the message. An mpsc send that is cancelled after it
//! wrote part of its message cannot reset the stream, which would drop earlier
//! messages that the receiver did not read yet. The next send writes the rest
//! of the frame with the marker for an aborted message instead, and the
//! receiver drops that message, so a cancelled send is not delivered, as for a
//! local channel. If the sender drops instead, the stream ends in the middle of
//! the message, and the receiver gets the end of the stream after the last
//! complete message, as for a sender that was dropped.
//!
//! The marker is compatible with earlier versions of irpc, which have no
//! marker. Such a version decodes the message and ignores the byte after it,
//! so it delivers a cancelled message that it gets. A frame without a marker
//! counts as a complete message.
//!
//! If an mpsc frame does not decode and ends with the marker for an aborted
//! message, the receiver treats it as aborted and drops it. This lets
//! cancellation discard a message even when the receiver rejects its contents.
//! An invalid frame from an older peer that ends with the same byte is also
//! dropped, instead of returning [`RpcRecvError::DecodeFailed`]. Valid
//! messages from older peers are unaffected.
//!
//! # Error codes
//!
//! irpc closes connections, and stops and resets streams, with the codes of
//! [`ErrorCode`]. irpc reserves the codes 0 to 15, and can add more codes in
//! this range in a later version.
//!
//! An application should use codes from 16 upward, so that a peer can tell its
//! codes apart from the codes of irpc. irpc does not check this. A code below
//! 16 works, but a later version of irpc can give it a different meaning.
//!
//! [`Client::rpc`]: crate::Client::rpc
use std::{
    borrow::BorrowMut,
    fmt::Debug,
    future::Future,
    io,
    marker::PhantomData,
    ops::Range,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll, ready},
};

use n0_error::{e, stack_error};
use n0_future::future::Boxed as BoxFuture;
use noq::ConnectionError;
/// The number type of the codes on streams and connections, see [`ErrorCode`].
pub use noq::VarInt;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use smallvec::{SmallVec, smallvec};
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tracing::{Instrument, debug, trace, warn};

#[cfg(feature = "iroh")]
pub(crate) mod iroh_impl;
pub(crate) mod noq_impl;

use crate::{
    LocalSender, RequestError, RpcMessage, Service,
    channel::{
        SendError,
        mpsc::{self, DynReceiver, DynSender},
        none::NoSender,
        oneshot,
    },
    util::{AsyncReadVarintExt, WriteVarintExt, now_or_never},
};

/// Default max message size (16 MiB).
///
/// The limit is on the encoded message after its size prefix. For an mpsc
/// message, it includes the byte that marks the end of the message, so the
/// largest mpsc message is one byte smaller than the largest oneshot message
/// or request.
pub const MAX_MESSAGE_SIZE: u64 = 1024 * 1024 * 16;

/// An error code on streams and connections.
///
/// irpc stops and resets streams, and closes connections, with the codes of
/// the named variants. A `Request` code is about the request, and a `Message`
/// code is about a channel message, see [Requests and channel messages]. A code
/// that the peer sent and that this version of irpc does not know is
/// [`Other`](Self::Other). An application should use codes from 16 upward, see
/// [Error codes].
///
/// Code 0 means no error, for example a normal close or a dropped receiver, so
/// it is never an `ErrorCode`. irpc reports it as a closed channel instead, for
/// example [`SendError::ReceiverClosed`].
///
/// # Examples
///
/// ```
/// use irpc::rpc::{ErrorCode, VarInt};
///
/// assert_eq!(
///     ErrorCode::from_code(VarInt::from_u32(4)),
///     Some(ErrorCode::RequestDecodeFailed)
/// );
/// assert_eq!(
///     ErrorCode::RequestDecodeFailed.to_code(),
///     VarInt::from_u32(4)
/// );
/// let code = VarInt::from_u32(401);
/// assert_eq!(ErrorCode::from_code(code), Some(ErrorCode::Other(code)));
/// assert_eq!(ErrorCode::from_code(VarInt::from_u32(0)), None);
/// ```
///
/// [Requests and channel messages]: self#requests-and-channel-messages
/// [Error codes]: self#error-codes
#[derive(Debug, derive_more::Display, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum ErrorCode {
    /// The client reset the stream of the request before the server read the request.
    ///
    /// For example, the client dropped the request before it wrote all of it.
    /// The server skips such a request, and resets its side of the stream with
    /// this code, see [Abandoned requests].
    ///
    /// [Abandoned requests]: self#abandoned-requests
    #[display("request aborted")]
    RequestAborted,
    /// The request is larger than [`MAX_MESSAGE_SIZE`].
    ///
    /// The server rejects such a request, see [Bad requests]. A client that
    /// finds its request too large resets the stream with this code.
    ///
    /// [Bad requests]: self#bad-requests
    #[display("request too large")]
    RequestMaxSizeExceeded,
    /// The request does not encode.
    ///
    /// A client that cannot encode its request resets the stream with this code.
    #[display("request does not encode")]
    RequestEncodeFailed,
    /// The request does not decode, or its stream ended before it was complete.
    ///
    /// The server rejects such a request, see [Bad requests].
    ///
    /// [Bad requests]: self#bad-requests
    #[display("request does not decode")]
    RequestDecodeFailed,
    /// A channel message is larger than [`MAX_MESSAGE_SIZE`].
    #[display("message too large")]
    MessageMaxSizeExceeded,
    /// A channel message does not encode.
    #[display("message does not encode")]
    MessageEncodeFailed,
    /// A channel message does not decode.
    #[display("message does not decode")]
    MessageDecodeFailed,
    /// The oneshot sender dropped its message before it wrote all of it.
    ///
    /// For example, the future of the send was dropped while it wrote a large
    /// message. The receiver reports this as
    /// [`oneshot::RecvError::SenderClosed`].
    #[display("message aborted")]
    MessageAborted,
    /// A code that this version of irpc does not know.
    ///
    /// This is a code of the application, see [Error codes], or a code that a
    /// newer version of irpc added. [`Self::from_code`] returns this only for a
    /// code that is not 0 and not the code of one of the variants above.
    ///
    /// [Error codes]: self#error-codes
    #[display("code {_0}")]
    Other(VarInt),
}

impl ErrorCode {
    /// Every error code that irpc knows, for [`Self::from_code`].
    const KNOWN: [Self; 8] = [
        Self::RequestAborted,
        Self::RequestMaxSizeExceeded,
        Self::RequestEncodeFailed,
        Self::RequestDecodeFailed,
        Self::MessageMaxSizeExceeded,
        Self::MessageEncodeFailed,
        Self::MessageDecodeFailed,
        Self::MessageAborted,
    ];

    /// Returns the code that streams and connections carry for this error code.
    pub const fn to_code(self) -> VarInt {
        match self {
            Self::RequestAborted => VarInt::from_u32(1),
            Self::RequestMaxSizeExceeded => VarInt::from_u32(2),
            Self::RequestEncodeFailed => VarInt::from_u32(3),
            Self::RequestDecodeFailed => VarInt::from_u32(4),
            Self::MessageMaxSizeExceeded => VarInt::from_u32(5),
            Self::MessageEncodeFailed => VarInt::from_u32(6),
            Self::MessageDecodeFailed => VarInt::from_u32(7),
            Self::MessageAborted => VarInt::from_u32(8),
            Self::Other(code) => code,
        }
    }

    /// Returns the error code for `code`, or `None` for code 0, which means no error.
    pub fn from_code(code: VarInt) -> Option<Self> {
        if code.into_inner() == 0 {
            return None;
        }
        let known = Self::KNOWN.into_iter().find(|c| c.to_code() == code);
        Some(known.unwrap_or(Self::Other(code)))
    }
}

/// How a connection was closed on purpose, by the peer or by this side.
///
/// A connection that failed, for example because the peer is gone, is a
/// `ConnectionFailed` error instead.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ConnectionClose {
    /// The peer closed the connection.
    Remote {
        /// The code of the close, or `None` for code 0, a normal close.
        code: Option<ErrorCode>,
        /// The reason that the peer gave.
        reason: Vec<u8>,
    },
    /// This side closed the connection.
    Local,
}

impl ConnectionClose {
    /// Returns the code that the peer closed the connection with, if it was not 0.
    pub fn code(&self) -> Option<ErrorCode> {
        match self {
            Self::Remote { code, .. } => *code,
            Self::Local => None,
        }
    }

    /// Returns how the connection was closed, or `None` if it failed.
    pub(crate) fn from_error(err: &ConnectionError) -> Option<Self> {
        match err {
            ConnectionError::ApplicationClosed(close) => Some(Self::Remote {
                code: ErrorCode::from_code(close.error_code),
                reason: close.reason.to_vec(),
            }),
            ConnectionError::LocallyClosed => Some(Self::Local),
            _ => None,
        }
    }

    /// Returns the kind of io error for this close, as noq uses for a [`ConnectionError`].
    pub(crate) fn io_kind(&self) -> io::ErrorKind {
        match self {
            Self::Remote { .. } => io::ErrorKind::ConnectionAborted,
            Self::Local => io::ErrorKind::NotConnected,
        }
    }

    /// Returns how the connection was closed, if `err` is from a stream of a closed connection.
    pub(crate) fn from_io(err: &io::Error) -> Option<Self> {
        let err = err.get_ref()?;
        if let Some(noq::ReadError::ConnectionLost(err)) = err.downcast_ref() {
            Self::from_error(err)
        } else if let Some(noq::WriteError::ConnectionLost(err)) = err.downcast_ref() {
            Self::from_error(err)
        } else {
            None
        }
    }
}

impl std::fmt::Display for ConnectionClose {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Remote { code, reason } => {
                write!(f, "closed by the peer")?;
                if let Some(code) = code {
                    write!(f, " ({code})")?;
                }
                if !reason.is_empty() {
                    write!(f, ": {}", String::from_utf8_lossy(reason))?;
                }
                Ok(())
            }
            Self::Local => write!(f, "closed by this side"),
        }
    }
}

/// Error when reading a request with [`read_request`].
///
/// For [`MaxSizeExceeded`], [`Malformed`], and [`DecodeFailed`], irpc stops
/// and resets the streams of the request with
/// [`ErrorCode::RequestMaxSizeExceeded`] or [`ErrorCode::RequestDecodeFailed`],
/// so its client gets an error. The connection is still usable, so a server
/// can read the next request.
///
/// [`MaxSizeExceeded`]: Self::MaxSizeExceeded
/// [`Malformed`]: Self::Malformed
/// [`DecodeFailed`]: Self::DecodeFailed
#[stack_error(derive, add_meta)]
#[non_exhaustive]
pub enum ReadRequestError {
    /// The request is larger than [`MAX_MESSAGE_SIZE`].
    #[error("Request too large")]
    MaxSizeExceeded,
    /// The request is not framed correctly.
    ///
    /// For example, the stream ended before the request was complete.
    #[error("Request malformed")]
    Malformed {
        #[error(std_err)]
        source: io::Error,
    },
    /// The request does not decode.
    ///
    /// For example, a newer client sent a request type that this server does
    /// not know.
    #[error("Request does not decode")]
    DecodeFailed {
        #[error(std_err)]
        source: postcard::Error,
    },
    /// The peer closed the connection with a code other than 0.
    ///
    /// A close with code 0, or by this side, is not an error: [`read_request`]
    /// returns `None`.
    #[error("Connection {close}")]
    ConnectionClosed { close: ConnectionClose },
    /// The connection failed, for example because it timed out.
    #[error("Connection failed")]
    ConnectionFailed {
        #[error(std_err)]
        source: ConnectionError,
    },
}

impl ReadRequestError {
    /// Returns the code that the peer closed the connection with, if any.
    pub fn remote_code(&self) -> Option<ErrorCode> {
        match self {
            Self::ConnectionClosed { close, .. } => close.code(),
            _ => None,
        }
    }

    /// Returns the irpc error code for a bad request, or `None` for a connection error.
    fn error_code(&self) -> Option<ErrorCode> {
        match self {
            Self::MaxSizeExceeded { .. } => Some(ErrorCode::RequestMaxSizeExceeded),
            Self::Malformed { .. } | Self::DecodeFailed { .. } => {
                Some(ErrorCode::RequestDecodeFailed)
            }
            Self::ConnectionClosed { .. } | Self::ConnectionFailed { .. } => None,
        }
    }
}

impl From<ReadRequestError> for io::Error {
    fn from(err: ReadRequestError) -> Self {
        match err {
            ReadRequestError::ConnectionFailed { source, .. } => source.into(),
            ReadRequestError::ConnectionClosed { ref close, .. } => {
                let kind = close.io_kind();
                io::Error::new(kind, err)
            }
            err => io::Error::new(io::ErrorKind::InvalidData, err),
        }
    }
}

/// Error of a remote sender that a local sender does not have.
///
/// A local sender only fails when the receiver is dropped. A remote sender can
/// also fail because of a message that does not encode or fit, because the
/// receiver stopped the stream with a code, or because of the connection.
/// [`SendError`] has the variant for the local case, and wraps this for the
/// rest.
///
/// If the peer rejects the request and also closes the connection, the close
/// usually arrives first, and the sender gets [`ConnectionClosed`] instead of
/// [`Remote`]. [`Self::remote_code`] returns the code in both cases.
///
/// [`SendError`]: crate::channel::SendError
/// [`Remote`]: Self::Remote
/// [`ConnectionClosed`]: Self::ConnectionClosed
#[stack_error(derive, add_meta)]
#[non_exhaustive]
pub enum RpcSendError {
    /// The message is larger than [`MAX_MESSAGE_SIZE`], so this sender closed the channel.
    #[error("Message too large")]
    MaxSizeExceeded,
    /// The message does not encode, so this sender closed the channel.
    #[error("Message does not encode")]
    EncodeFailed {
        #[error(std_err)]
        source: postcard::Error,
    },
    /// The receiver stopped the stream with a code other than 0, and the channel is closed.
    ///
    /// The `code` says why, see [`ErrorCode`], and the [`rpc`](self) module for
    /// an overview. A stop with code 0 is [`SendError::ReceiverClosed`].
    ///
    /// [`SendError::ReceiverClosed`]: crate::channel::SendError::ReceiverClosed
    #[error("Receiver stopped the stream: {code}")]
    Remote { code: ErrorCode },
    /// The connection was closed on purpose, by the peer or by this side.
    #[error("Connection {close}")]
    ConnectionClosed { close: ConnectionClose },
    /// This sender closed the channel after an earlier message did not encode or was too large.
    #[error("Sender closed after an earlier error")]
    Closed,
    /// The connection failed, for example because it was lost or timed out.
    #[error("Connection failed")]
    ConnectionFailed {
        #[error(std_err)]
        source: io::Error,
    },
}

impl RpcSendError {
    /// Returns the code that the peer sent, if any.
    ///
    /// This is the code of a [`Remote`](Self::Remote) error, or the code that
    /// the peer closed the connection with.
    pub fn remote_code(&self) -> Option<ErrorCode> {
        match self {
            Self::Remote { code, .. } => Some(*code),
            Self::ConnectionClosed { close, .. } => close.code(),
            _ => None,
        }
    }
}

impl From<RpcSendError> for io::Error {
    fn from(e: RpcSendError) -> Self {
        match e {
            RpcSendError::ConnectionFailed { source, .. } => source,
            RpcSendError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
            RpcSendError::ConnectionClosed { ref close, .. } => {
                let kind = close.io_kind();
                io::Error::new(kind, e)
            }
            RpcSendError::Closed { .. } => io::Error::new(io::ErrorKind::BrokenPipe, e),
            _ => io::Error::new(io::ErrorKind::InvalidData, e),
        }
    }
}

/// Error of a remote receiver that a local receiver does not have.
///
/// A local receiver only ends when the sender is dropped. A remote receiver can
/// also fail because of a message that does not decode or fit, because the
/// sender reset the stream with a code, or because of the connection. Each
/// receive error has the variant for the local case, and wraps this for the
/// rest.
///
/// If the server rejects the request and also closes the connection, the close
/// usually arrives first, and the receiver gets [`ConnectionClosed`] instead of
/// [`Remote`]. [`Self::remote_code`] returns the code in both cases.
///
/// [`Remote`]: Self::Remote
/// [`ConnectionClosed`]: Self::ConnectionClosed
#[stack_error(derive, add_meta)]
#[non_exhaustive]
pub enum RpcRecvError {
    /// The message is larger than [`MAX_MESSAGE_SIZE`], so this receiver closed the channel.
    #[error("Message too large")]
    MaxSizeExceeded,
    /// The message does not decode, so this receiver closed the channel.
    #[error("Message does not decode")]
    DecodeFailed {
        #[error(std_err)]
        source: postcard::Error,
    },
    /// The sender reset the stream, and the channel is closed.
    ///
    /// The `code` says why, see [`ErrorCode`], and the [`rpc`](self) module for
    /// an overview. It is `None` for code 0, where that is not the same as a
    /// local channel whose sender was dropped: a reset with code 0 can drop
    /// messages that the receiver did not read yet. This is only the case for
    /// an mpsc receiver; a oneshot receiver gets
    /// [`SenderClosed`](crate::channel::oneshot::RecvError::SenderClosed) for code 0.
    #[error("Sender reset the stream{}", code.map(|code| format!(": {code}")).unwrap_or_default())]
    Remote { code: Option<ErrorCode> },
    /// The connection was closed on purpose, by the peer or by this side.
    #[error("Connection {close}")]
    ConnectionClosed { close: ConnectionClose },
    /// The connection failed, for example because it was lost or timed out.
    #[error("Connection failed")]
    ConnectionFailed {
        #[error(std_err)]
        source: io::Error,
    },
}

impl RpcRecvError {
    /// Returns the code that the peer sent, if any.
    ///
    /// This is the code of a [`Remote`](Self::Remote) error, or the code that
    /// the peer closed the connection with.
    pub fn remote_code(&self) -> Option<ErrorCode> {
        match self {
            Self::Remote { code, .. } => *code,
            Self::ConnectionClosed { close, .. } => close.code(),
            _ => None,
        }
    }
}

/// Converts an io error from reading the stream of a remote channel.
///
/// A reset becomes [`RpcRecvError::Remote`], the close of the connection becomes
/// [`RpcRecvError::ConnectionClosed`], and any other io error becomes [`RpcRecvError::ConnectionFailed`].
impl From<io::Error> for RpcRecvError {
    fn from(err: io::Error) -> Self {
        if let Some(code) = reset_code(&err) {
            return e!(RpcRecvError::Remote { code });
        }
        if let Some(close) = ConnectionClose::from_io(&err) {
            return e!(RpcRecvError::ConnectionClosed { close });
        }
        e!(RpcRecvError::ConnectionFailed, err)
    }
}

impl From<RpcRecvError> for io::Error {
    fn from(e: RpcRecvError) -> Self {
        match e {
            RpcRecvError::ConnectionFailed { source, .. } => source,
            RpcRecvError::Remote { .. } => io::Error::new(io::ErrorKind::ConnectionReset, e),
            RpcRecvError::ConnectionClosed { ref close, .. } => {
                let kind = close.io_kind();
                io::Error::new(kind, e)
            }
            _ => io::Error::new(io::ErrorKind::InvalidData, e),
        }
    }
}

/// Trait to abstract over a client connection to a remote service.
///
/// This isn't really that much abstracted, since the result of open_bi must
/// still be a noq::SendStream and noq::RecvStream. This is just so we
/// can have different connection implementations for normal noq connections,
/// iroh connections, and possibly noq connections with disabled encryption
/// for performance.
pub trait RemoteConnection: Send + Sync + Debug + 'static {
    /// Boxed clone so the trait is dynable.
    fn clone_boxed(&self) -> Box<dyn RemoteConnection>;

    /// Open a bidirectional stream to the remote service.
    fn open_bi(
        &self,
    ) -> BoxFuture<std::result::Result<(noq::SendStream, noq::RecvStream), RequestError>>;

    /// Returns whether 0-RTT data was rejected by the server.
    ///
    /// For connections that were fully authenticated before allowing to send any data, this should return `false`.
    fn zero_rtt_rejected(&self) -> BoxFuture<bool>;
}

/// A connection to a remote service that can be used to send the initial message.
///
/// If it drops before the message is written, it resets the stream with
/// [`ErrorCode::RequestAborted`].
#[derive(Debug)]
pub struct RemoteSender<S>(ResetOnDrop, noq::RecvStream, std::marker::PhantomData<S>);

/// A send stream that resets with its code if it drops before [`Self::into_inner`].
///
/// It owns the stream, or borrows it for a single write.
#[derive(Debug)]
struct ResetOnDrop<S: BorrowMut<noq::SendStream> = noq::SendStream> {
    send: Option<S>,
    /// The code that the stream resets with on drop.
    code: ErrorCode,
}

impl<S: BorrowMut<noq::SendStream>> ResetOnDrop<S> {
    fn new(send: S, code: ErrorCode) -> Self {
        Self {
            send: Some(send),
            code,
        }
    }

    fn get_mut(&mut self) -> &mut noq::SendStream {
        self.send
            .as_mut()
            .expect("only `into_inner` and `reset` take the stream")
            .borrow_mut()
    }

    fn into_inner(mut self) -> S {
        self.send
            .take()
            .expect("only `into_inner` and `reset` take the stream")
    }

    /// Resets the stream with `code`, instead of the code of this guard on drop.
    fn reset(&mut self, code: ErrorCode) {
        if let Some(mut send) = self.send.take() {
            send.borrow_mut().reset(code.to_code()).ok();
        }
    }
}

impl<S: BorrowMut<noq::SendStream>> Drop for ResetOnDrop<S> {
    fn drop(&mut self) {
        if let Some(send) = &mut self.send {
            send.borrow_mut().reset(self.code.to_code()).ok();
        }
    }
}

/// The inline size of the buffers that frames are encoded and received in.
///
/// Frames up to this size, length prefix included, need no allocation. A larger
/// frame moves a buffer to the heap. Buffers that live as long as a stream move
/// back inline once they are empty again, so that one large message does not
/// pin its allocation for the stream's life.
const INLINE_FRAME_SIZE: usize = 128;

/// Serialize a message for sending over the wire.
///
/// When `S::SPAN_PROPAGATION` is true, the message is wrapped in a tuple with
/// the context of `span`: `(Option<SpanContextCarrier>, msg)`.
/// When false, the message is serialized directly.
fn prepare_write<S: Service>(
    msg: impl Into<S>,
    span: &tracing::Span,
) -> Result<SmallVec<[u8; INLINE_FRAME_SIZE]>, EncodeError> {
    let msg = msg.into();
    let mut buf = SmallVec::new();
    let res = if S::SPAN_PROPAGATION {
        // Include span context in wire format
        let span_ctx = crate::span_propagation::SpanContextCarrier::from_span_or_current(span);
        encode(&(span_ctx, msg), &mut buf)
    } else {
        // Original wire format without span context
        encode(&msg, &mut buf)
    };
    res?;
    Ok(buf)
}

/// Why a message did not encode.
enum EncodeError {
    /// The message is larger than [`MAX_MESSAGE_SIZE`].
    MaxSizeExceeded,
    /// The message does not encode.
    Failed(postcard::Error),
}

/// Encodes `value` with its size prefix into `buf`.
fn encode(
    value: &impl Serialize,
    buf: &mut SmallVec<[u8; INLINE_FRAME_SIZE]>,
) -> Result<(), EncodeError> {
    let size = postcard::experimental::serialized_size(value).map_err(EncodeError::Failed)? as u64;
    if size > MAX_MESSAGE_SIZE {
        return Err(EncodeError::MaxSizeExceeded);
    }
    buf.clear();
    buf.write_varint_u64(size)
        .expect("a buffer does not fail to write");
    postcard::to_io(value, buf).map_err(EncodeError::Failed)?;
    Ok(())
}

/// Encodes the channel message `value` with its size prefix into `buf`.
///
/// Resets `send` if `value` is too large or does not encode, so the receiver
/// gets an error instead of the end of the stream.
fn encode_message(
    value: &impl Serialize,
    buf: &mut SmallVec<[u8; INLINE_FRAME_SIZE]>,
    send: &mut noq::SendStream,
) -> Result<(), SendError> {
    encode(value, buf).map_err(|err| {
        let (code, err) = match err {
            EncodeError::MaxSizeExceeded => (
                ErrorCode::MessageMaxSizeExceeded,
                e!(RpcSendError::MaxSizeExceeded),
            ),
            EncodeError::Failed(source) => (
                ErrorCode::MessageEncodeFailed,
                e!(RpcSendError::EncodeFailed, source),
            ),
        };
        send.reset(code.to_code()).ok();
        err.into()
    })
}

impl<S: Service> RemoteSender<S> {
    pub fn new(send: noq::SendStream, recv: noq::RecvStream) -> Self {
        Self(
            ResetOnDrop::new(send, ErrorCode::RequestAborted),
            recv,
            PhantomData,
        )
    }

    /// Writes the request `msg` in its span from [`Service::new_request_span`],
    /// and returns the streams of the request.
    ///
    /// If `msg` is too large or does not encode, this resets the stream with the
    /// code of the error.
    pub async fn write(
        self,
        msg: impl Into<S>,
    ) -> std::result::Result<(noq::SendStream, noq::RecvStream), SendError> {
        let (send, recv, _span) = self.write_in_span(msg).await?;
        Ok((send, recv))
    }

    /// Writes the request like [`Self::write`], and also returns its span.
    pub(crate) async fn write_in_span(
        mut self,
        msg: impl Into<S>,
    ) -> std::result::Result<(noq::SendStream, noq::RecvStream, tracing::Span), SendError> {
        let (span, buf) = self.prepare(msg)?;
        let (send, recv) = self.write_raw(&buf).instrument(span.clone()).await?;
        Ok((send, recv, span))
    }

    /// Returns the index of the stream, which the server also uses as the request id.
    pub(crate) fn req_id(&self) -> u64 {
        self.1.id().index()
    }

    /// Creates the span of the request, and encodes the request.
    ///
    /// With span propagation, the request carries the context of this span, or
    /// of the current span if the tracing backend does not record this span.
    /// If `msg` is too large or does not encode, this resets the stream with
    /// the code of the error, so the server learns why the request is missing.
    pub(crate) fn prepare(
        &mut self,
        msg: impl Into<S>,
    ) -> std::result::Result<(tracing::Span, SmallVec<[u8; INLINE_FRAME_SIZE]>), SendError> {
        let msg = msg.into();
        let span = msg.new_request_span(self.req_id());
        let buf = prepare_write::<S>(msg, &span).map_err(|err| {
            let (code, err) = match err {
                EncodeError::MaxSizeExceeded => (
                    ErrorCode::RequestMaxSizeExceeded,
                    e!(RpcSendError::MaxSizeExceeded),
                ),
                EncodeError::Failed(source) => (
                    ErrorCode::RequestEncodeFailed,
                    e!(RpcSendError::EncodeFailed, source),
                ),
            };
            self.0.reset(code);
            SendError::from(err)
        })?;
        Ok((span, buf))
    }

    pub(crate) async fn write_raw(
        self,
        buf: &[u8],
    ) -> std::result::Result<(noq::SendStream, noq::RecvStream), SendError> {
        let RemoteSender(mut send, recv, _) = self;
        send.get_mut().write_all(buf).await?;
        debug!("request sent");
        Ok((send.into_inner(), recv))
    }
}

impl<T: DeserializeOwned> From<noq::RecvStream> for oneshot::Receiver<T> {
    fn from(mut read: noq::RecvStream) -> Self {
        // A client creates the receiver in the request span, and logs the response in it.
        let span = tracing::Span::current();
        let fut = async move {
            // A sender that is dropped without sending finishes the stream.
            let Some(size) = read_size(&mut read).await? else {
                return Err(e!(oneshot::RecvError::SenderClosed));
            };
            if size > MAX_MESSAGE_SIZE {
                read.stop(ErrorCode::MessageMaxSizeExceeded.to_code()).ok();
                return Err(e!(RpcRecvError::MaxSizeExceeded).into());
            }
            let mut buf = vec![0; size as usize];
            AsyncReadExt::read_exact(&mut read, &mut buf).await?;
            let msg: T = postcard::from_bytes(&buf).map_err(|err| {
                read.stop(ErrorCode::MessageDecodeFailed.to_code()).ok();
                oneshot::RecvError::from(e!(RpcRecvError::DecodeFailed, err))
            })?;
            span.in_scope(|| debug!("response received"));
            Ok(msg)
        };
        oneshot::Receiver::from(|| fut)
    }
}

/// Returns the noq read error in `err`, if the stream failed.
fn read_error(err: &io::Error) -> Option<&noq::ReadError> {
    err.get_ref()?.downcast_ref()
}

/// Returns the noq write error in `err`, if the stream failed.
fn write_error(err: &io::Error) -> Option<&noq::WriteError> {
    err.get_ref()?.downcast_ref()
}

impl SendError {
    /// Returns whether the server rejected 0-RTT while this client wrote the request.
    pub(crate) fn is_zero_rtt_rejected(&self) -> bool {
        match self {
            Self::Rpc {
                source: RpcSendError::ConnectionFailed { source, .. },
                ..
            } => matches!(write_error(source), Some(noq::WriteError::ZeroRttRejected)),
            _ => false,
        }
    }
}

/// Reads the size prefix of a message, or returns `None` at the end of the stream.
///
/// A size prefix that does not fit in a `u64` is larger than
/// [`MAX_MESSAGE_SIZE`], so this returns `u64::MAX` for it.
async fn read_size(recv: &mut noq::RecvStream) -> io::Result<Option<u64>> {
    match recv.read_varint_u64().await {
        // The only error of the varint itself: all others are from the stream.
        Err(err) if err.kind() == io::ErrorKind::InvalidData && read_error(&err).is_none() => {
            Ok(Some(u64::MAX))
        }
        res => res,
    }
}

/// Returns whether `err` is the end of a stream in the middle of a message.
///
/// The reads of a message return an `UnexpectedEof` error that is not from noq
/// when the stream finishes early.
pub(crate) fn is_truncated(err: &io::Error) -> bool {
    err.kind() == io::ErrorKind::UnexpectedEof && read_error(err).is_none()
}

/// Returns the code that the sender reset the stream with, if it did.
///
/// The inner `None` is code 0.
pub(crate) fn reset_code(err: &io::Error) -> Option<Option<ErrorCode>> {
    match read_error(err)? {
        noq::ReadError::Reset(code) => Some(ErrorCode::from_code(*code)),
        _ => None,
    }
}

impl From<noq::RecvStream> for crate::channel::none::NoReceiver {
    fn from(read: noq::RecvStream) -> Self {
        drop(read);
        Self
    }
}

impl<T: RpcMessage> From<noq::RecvStream> for mpsc::Receiver<T> {
    fn from(read: noq::RecvStream) -> Self {
        mpsc::Receiver::from_boxed(Box::new(NoqReceiver {
            recv: read,
            span: tracing::Span::current(),
            received: 0,
            finished: false,
            buf: smallvec![0; INLINE_FRAME_SIZE],
            end: 0,
            _marker: PhantomData,
        }))
    }
}

impl From<noq::SendStream> for NoSender {
    fn from(write: noq::SendStream) -> Self {
        let _ = write;
        NoSender
    }
}

impl<T: RpcMessage> From<noq::SendStream> for oneshot::Sender<T> {
    fn from(writer: noq::SendStream) -> Self {
        oneshot::Sender::from_boxed(Box::new(NoqOneshotSender {
            writer,
            // A server creates the sender in the request span, and logs the response in it.
            span: tracing::Span::current(),
        }))
    }
}

#[derive(Debug)]
struct NoqOneshotSender {
    writer: noq::SendStream,
    span: tracing::Span,
}

impl<T: RpcMessage> oneshot::DynSender<T> for NoqOneshotSender {
    fn send(self: Box<Self>, value: T) -> BoxFuture<Result<(), SendError>> {
        let Self { mut writer, span } = *self;
        Box::pin(async move {
            // write via a small buffer to avoid allocation for small values
            let mut buf = SmallVec::new();
            encode_message(&value, &mut buf, &mut writer)?;
            // If this send drops in the middle of the message, reset the
            // stream, so the receiver gets an error instead of a part of it.
            let mut guard = ResetOnDrop::new(&mut writer, ErrorCode::MessageAborted);
            let res = guard.get_mut().write_all(&buf).await;
            guard.into_inner();
            res?;
            span.in_scope(|| debug!("response sent"));
            Ok(())
        })
    }

    fn is_rpc(&self) -> bool {
        true
    }
}

impl<T: RpcMessage> From<noq::SendStream> for mpsc::Sender<T> {
    fn from(write: noq::SendStream) -> Self {
        mpsc::Sender::from_boxed(Arc::new(NoqSender(tokio::sync::Mutex::new(
            NoqSenderState::Open(NoqSenderInner {
                send: write,
                buffer: SmallVec::new(),
                written: 0,
                span: tracing::Span::current(),
                sent: 0,
                _marker: PhantomData,
            }),
        ))))
    }
}

/// The receiver of a remote [`mpsc`] channel.
///
/// It logs the first message and the end of the stream in the span in which it was created.
struct NoqReceiver<T> {
    recv: noq::RecvStream,
    span: tracing::Span,
    received: u64,
    /// EOF or an error has ended the channel; later polls return `Ok(None)`.
    finished: bool,
    /// Zero-initialized space the receiver reads into, so the most it reads ahead.
    buf: SmallVec<[u8; INLINE_FRAME_SIZE]>,
    /// `buf[..end]` holds the bytes read but not yet consumed as a message.
    end: usize,
    _marker: std::marker::PhantomData<T>,
}

/// What the buffered bytes of a length-prefixed stream hold next.
#[derive(Debug, PartialEq)]
enum Frame {
    /// The buffer ends before the frame does.
    ///
    /// `frame_len` is the length of the whole frame, varint prefix included,
    /// or `None` while the varint prefix itself is still incomplete.
    Partial { frame_len: Option<usize> },
    /// The buffer starts with a whole frame.
    ///
    /// `body` is the range of the frame's content after the varint prefix;
    /// it ends where the frame does.
    Whole { body: Range<usize> },
    /// The length prefix exceeds [`MAX_MESSAGE_SIZE`].
    TooLarge,
}

/// How an mpsc message ends: the last part of its frame, after the message.
///
/// Each variant encodes as one byte, so a sender can mark a message that it
/// already started as aborted by encoding the last byte again.
#[derive(Debug, Serialize, Deserialize)]
enum MessageEnd {
    /// The receiver delivers the message.
    Complete,
    /// The send of the message was cancelled, so the receiver drops it.
    Aborted,
}

/// Decodes an mpsc frame into its message, or `None` if the send of the message was cancelled.
///
/// A frame from an earlier version of irpc ends after the message, and counts as complete.
///
/// The content of a cancelled message does not matter, so it does not have to
/// decode: this receiver may validate more than the sender did. A frame that
/// does not decode and ends with the marker for an aborted message is dropped.
/// A frame from an earlier version that does not decode and happens to end
/// with the same byte is dropped as well, instead of failing to decode.
fn decode_frame<T: DeserializeOwned>(buf: &[u8]) -> postcard::Result<Option<T>> {
    let (msg, rest) = match postcard::take_from_bytes(buf) {
        Ok(decoded) => decoded,
        Err(_) if ends_aborted(buf) => return Ok(None),
        Err(err) => return Err(err),
    };
    let end = match rest {
        [] => MessageEnd::Complete,
        [_] => postcard::from_bytes(rest)?,
        _ if ends_aborted(buf) => return Ok(None),
        _ => return Err(postcard::Error::DeserializeBadEncoding),
    };
    Ok(match end {
        MessageEnd::Complete => Some(msg),
        MessageEnd::Aborted => None,
    })
}

/// Returns whether `buf` ends with the marker for an aborted message.
fn ends_aborted(buf: &[u8]) -> bool {
    let last = buf.last().map(std::slice::from_ref).unwrap_or_default();
    matches!(postcard::from_bytes(last), Ok(MessageEnd::Aborted))
}

/// Finds the next length-prefixed message at the start of `buf`, if it is all there.
fn parse_frame(buf: &[u8]) -> io::Result<Frame> {
    // The length prefix is a postcard varint, so postcard parses it.
    let (size, rest) = match postcard::take_from_bytes::<u64>(buf) {
        Ok(parsed) => parsed,
        Err(postcard::Error::DeserializeUnexpectedEnd) => {
            return Ok(Frame::Partial { frame_len: None });
        }
        // A size that does not fit in a u64 is larger than MAX_MESSAGE_SIZE.
        Err(postcard::Error::DeserializeBadVarint) => return Ok(Frame::TooLarge),
        Err(err) => return Err(io::Error::new(io::ErrorKind::InvalidData, err)),
    };
    let prefix = buf.len() - rest.len();
    if size > MAX_MESSAGE_SIZE {
        return Ok(Frame::TooLarge);
    }
    let end = prefix + size as usize;
    if buf.len() < end {
        return Ok(Frame::Partial {
            frame_len: Some(end),
        });
    }
    Ok(Frame::Whole { body: prefix..end })
}

impl<T> NoqReceiver<T> {
    /// Handles the end of the stream, and returns `None`.
    ///
    /// The stream can end in the middle of a message if the send of that
    /// message was cancelled. The sender never finished that send, so this is
    /// the same as a sender that was dropped after its last complete message.
    fn finish(&mut self) -> Option<T> {
        if self.end > 0 {
            self.span.in_scope(|| {
                debug!("stream ended in the middle of a message, the send was cancelled")
            });
        }
        self.end = 0;
        if self.buf.spilled() {
            self.buf.truncate(INLINE_FRAME_SIZE);
            self.buf.shrink_to_fit();
        }
        if !self.finished {
            self.finished = true;
            let received = self.received;
            self.span.in_scope(|| debug!(received, "stream finished"));
        }
        None
    }
}

impl<T> Debug for NoqReceiver<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NoqReceiver").finish()
    }
}

impl<T: RpcMessage> NoqReceiver<T> {
    fn poll_recv_inner(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<Result<Option<T>, mpsc::RecvError>> {
        loop {
            match parse_frame(&self.buf[..self.end])? {
                Frame::Whole { body } => {
                    let msg = decode_frame(&self.buf[body.clone()]);
                    self.buf.copy_within(body.end..self.end, 0);
                    self.end -= body.end;
                    // For large frames, we resized the buffer to exactly the frame size.
                    // After consuming the frame, the buffer is empty and we can
                    // move it back inline, releasing the heap storage.
                    if self.end == 0 && self.buf.spilled() {
                        self.buf.truncate(INLINE_FRAME_SIZE);
                        self.buf.shrink_to_fit();
                    }
                    let msg = msg.map_err(|err| {
                        self.recv
                            .stop(ErrorCode::MessageDecodeFailed.to_code())
                            .ok();
                        mpsc::RecvError::from(e!(RpcRecvError::DecodeFailed, err))
                    })?;
                    let Some(msg) = msg else {
                        self.span
                            .in_scope(|| debug!("dropped a message whose send was cancelled"));
                        continue;
                    };
                    self.received += 1;
                    if self.received == 1 {
                        self.span.in_scope(|| debug!("first message received"));
                    }
                    return Poll::Ready(Ok(Some(msg)));
                }
                Frame::TooLarge => {
                    self.recv
                        .stop(ErrorCode::MessageMaxSizeExceeded.to_code())
                        .ok();
                    return Poll::Ready(Err(e!(RpcRecvError::MaxSizeExceeded).into()));
                }
                Frame::Partial { frame_len } => {
                    // The frame starts at 0, so the buffer must hold all of it.
                    if let Some(len) = frame_len
                        && self.buf.len() < len
                    {
                        self.buf.resize(len, 0);
                    }
                    let mut read = ReadBuf::new(&mut self.buf[self.end..]);
                    ready!(Pin::new(&mut self.recv).poll_read(cx, &mut read))?;
                    let n = read.filled().len();
                    self.end += n;
                    if n == 0 {
                        return Poll::Ready(Ok(self.finish()));
                    }
                }
            }
        }
    }
}

impl<T: RpcMessage> DynReceiver<T> for NoqReceiver<T> {
    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<T>, mpsc::RecvError>> {
        if self.finished {
            return Poll::Ready(Ok(None));
        }
        let result = self.poll_recv_inner(cx);
        if matches!(&result, Poll::Ready(Err(_))) {
            // The error ends the channel, including any messages read ahead.
            self.finished = true;
            self.end = 0;
            clear_frame(&mut self.buf);
        }
        result
    }
}

impl<T> Drop for NoqReceiver<T> {
    fn drop(&mut self) {}
}

/// Empties a frame buffer that lives as long as a stream, moving it back inline.
fn clear_frame(buf: &mut SmallVec<[u8; INLINE_FRAME_SIZE]>) {
    buf.clear();
    if buf.spilled() {
        buf.shrink_to_fit();
    }
}

struct NoqSenderInner<T> {
    send: noq::SendStream,
    /// The encoded message that is being sent.
    buffer: SmallVec<[u8; INLINE_FRAME_SIZE]>,
    /// How many bytes of `buffer` the stream has taken.
    ///
    /// A send that is cancelled leaves its message here. If none of it was
    /// written, the next send drops it. Otherwise the next send writes the
    /// rest first, ending with [`MessageEnd::Aborted`], so that the stream
    /// stays a sequence of whole messages and the receiver drops this one. If
    /// the sender drops instead, the stream ends in the middle of the message.
    written: usize,
    /// The span in which the sender was created, for the logs of the first message and the end.
    span: tracing::Span,
    sent: u64,
    _marker: std::marker::PhantomData<T>,
}

impl<T: RpcMessage> NoqSenderInner<T> {
    /// Counts a sent message, and logs the first one.
    fn record_sent(&mut self) {
        self.sent += 1;
        if self.sent == 1 {
            self.span.in_scope(|| debug!("first message sent"));
        }
    }

    /// Writes the rest of the message in `buffer`.
    ///
    /// Cancel safe: the progress is in `written`.
    async fn write_rest(&mut self) -> Result<(), SendError> {
        while self.written < self.buffer.len() {
            self.written += self.send.write(&self.buffer[self.written..]).await?;
        }
        clear_frame(&mut self.buffer);
        self.written = 0;
        Ok(())
    }

    /// Prepares the buffer for a new message.
    ///
    /// Drops a message that a cancelled send did not start. A message that a
    /// cancelled send started is written to the end, but marked as aborted, so
    /// that the receiver drops it.
    async fn prepare(&mut self) -> Result<(), SendError> {
        if self.written > 0 {
            let last = self.buffer.len() - 1;
            postcard::to_slice(&MessageEnd::Aborted, &mut self.buffer[last..])
                .expect("the end of a message is one byte");
            self.write_rest().await?;
        }
        clear_frame(&mut self.buffer);
        Ok(())
    }

    /// Encodes `value` into the buffer, as a message that the receiver delivers.
    fn encode(&mut self, value: &T) -> Result<(), SendError> {
        let frame = (value, MessageEnd::Complete);
        encode_message(&frame, &mut self.buffer, &mut self.send)
    }

    async fn send(&mut self, value: T) -> Result<(), SendError> {
        self.prepare().await?;
        self.encode(&value)?;
        self.write_rest().await?;
        self.record_sent();
        Ok(())
    }

    async fn try_send(&mut self, value: T) -> Result<bool, SendError> {
        // A started message goes first: if it cannot, this one cannot either.
        let Some(res) = now_or_never(self.prepare()) else {
            return Ok(false);
        };
        res?;
        self.encode(&value)?;
        let Some(n) = now_or_never(self.send.write(&self.buffer)) else {
            clear_frame(&mut self.buffer);
            return Ok(false);
        };
        self.written = n?;
        self.write_rest().await?;
        self.record_sent();
        Ok(true)
    }
}

// A sender has one state, so the size of the open state costs nothing.
#[allow(clippy::large_enum_variant)]
enum NoqSenderState<T> {
    Open(NoqSenderInner<T>),
    Closed(CloseReason),
}

/// Why a remote sender is closed, so that later sends fail with the same error.
#[derive(Debug, Clone)]
enum CloseReason {
    /// The receiver closed the channel.
    ReceiverClosed,
    /// The receiver stopped the stream with a code.
    Remote(ErrorCode),
    /// The connection was closed on purpose.
    ConnectionClosed(ConnectionClose),
    /// The stream failed, for example because the connection was lost.
    Io(noq::WriteError),
    /// This sender closed the channel after an error.
    Other,
}

impl CloseReason {
    fn error(self) -> SendError {
        match self {
            Self::ReceiverClosed => e!(SendError::ReceiverClosed),
            Self::Remote(code) => e!(RpcSendError::Remote { code }).into(),
            Self::ConnectionClosed(close) => e!(RpcSendError::ConnectionClosed { close }).into(),
            Self::Io(err) => e!(RpcSendError::ConnectionFailed, err.into()).into(),
            Self::Other => e!(RpcSendError::Closed).into(),
        }
    }
}

impl<T: RpcMessage> NoqSenderState<T> {
    /// Runs `f` on the open sender, and closes the sender if it fails.
    ///
    /// If the future is dropped, the progress of the send stays in the sender.
    async fn run<R>(
        &mut self,
        f: impl AsyncFnOnce(&mut NoqSenderInner<T>) -> Result<R, SendError>,
    ) -> Result<R, SendError> {
        let sender = match self {
            Self::Open(sender) => sender,
            Self::Closed(reason) => return Err(reason.clone().error()),
        };
        let res = f(sender).await;
        match &res {
            Ok(_) => {}
            Err(SendError::ReceiverClosed { .. }) => {
                *self = Self::Closed(CloseReason::ReceiverClosed)
            }
            Err(SendError::Rpc { source, .. }) => {
                *self = Self::Closed(match source {
                    RpcSendError::Remote { code, .. } => CloseReason::Remote(*code),
                    RpcSendError::ConnectionClosed { close, .. } => {
                        CloseReason::ConnectionClosed(close.clone())
                    }
                    RpcSendError::ConnectionFailed { source, .. } => match write_error(source) {
                        Some(err) => CloseReason::Io(err.clone()),
                        None => CloseReason::Other,
                    },
                    _ => CloseReason::Other,
                })
            }
        }
        res
    }
}

struct NoqSender<T>(tokio::sync::Mutex<NoqSenderState<T>>);

impl<T> Drop for NoqSender<T> {
    /// Logs the end of the stream, which the send stream finishes when it drops.
    ///
    /// A sender that failed is closed, and logs nothing: the failed send returned the error.
    fn drop(&mut self) {
        if let NoqSenderState::Open(sender) = self.0.get_mut() {
            let sent = sender.sent;
            sender.span.in_scope(|| debug!(sent, "stream finished"));
        }
    }
}

impl<T> Debug for NoqSender<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("NoqSender").finish()
    }
}

impl<T: RpcMessage> DynSender<T> for NoqSender<T> {
    fn send(&self, value: T) -> Pin<Box<dyn Future<Output = Result<(), SendError>> + Send + '_>> {
        Box::pin(async {
            let mut state = self.0.lock().await;
            state.run(async |sender| sender.send(value).await).await
        })
    }

    fn try_send(
        &self,
        value: T,
    ) -> Pin<Box<dyn Future<Output = Result<bool, SendError>> + Send + '_>> {
        Box::pin(async {
            let mut state = self.0.lock().await;
            state.run(async |sender| sender.try_send(value).await).await
        })
    }

    fn closed(&self) -> Pin<Box<dyn Future<Output = ()> + Send + Sync + '_>> {
        Box::pin(async {
            // Wait without the lock, so that other clones can send meanwhile.
            let stopped = match &*self.0.lock().await {
                NoqSenderState::Open(sender) => sender.send.stopped(),
                NoqSenderState::Closed(_) => return,
            };
            stopped.await.ok();
        })
    }

    fn is_rpc(&self) -> bool {
        true
    }
}

/// The function inside a [`Handler`].
type HandlerFn<S> = Arc<
    dyn Fn(S, noq::RecvStream, noq::SendStream) -> BoxFuture<std::result::Result<(), HandlerError>>
        + Send
        + Sync
        + 'static,
>;

/// The error that a handler function returns to close the connection.
///
/// Currently the only constructor is [`Self::close_connection`]. The server
/// then closes the connection.
#[derive(Debug)]
pub struct HandlerError {
    inner: HandlerErrorInner,
}

#[derive(Debug)]
enum HandlerErrorInner {
    CloseConnection { code: VarInt, reason: Vec<u8> },
}

impl std::error::Error for HandlerError {}

impl std::fmt::Display for HandlerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.inner {
            HandlerErrorInner::CloseConnection { code, reason } => write!(
                f,
                "handler asked to close the connection with code {code}: {}",
                String::from_utf8_lossy(reason)
            ),
        }
    }
}

impl HandlerError {
    /// Creates a [`HandlerError`] that closes the connection with `code` and `reason`.
    ///
    /// When a handler function returns this error, the server closes the connection
    /// immediately and reads no more requests from it. This also ends the streams of
    /// all other requests on the connection, including streams that a handler moved
    /// into a spawned task.
    ///
    /// The code should be 16 or higher, see [Error codes]. irpc
    /// closes a connection with code 0 for a normal close, which
    /// [`Handler::from_sender`] also uses when its receiver is gone. For a bad
    /// request, irpc closes the connection with [`ErrorCode::RequestMaxSizeExceeded`]
    /// or [`ErrorCode::RequestDecodeFailed`].
    ///
    /// # Examples
    ///
    /// ```
    /// use irpc::{
    ///     channel::oneshot,
    ///     rpc::{Handler, HandlerError, VarInt},
    ///     rpc_requests,
    /// };
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[rpc_requests(message = AuthMessage)]
    /// #[derive(Debug, Serialize, Deserialize)]
    /// enum AuthProtocol {
    ///     #[rpc(tx = oneshot::Sender<()>)]
    ///     #[wrap(Auth)]
    ///     Auth(String),
    /// }
    ///
    /// let handler = Handler::<AuthProtocol>::sequential(|msg| async move {
    ///     let AuthMessage::Auth(msg) = msg;
    ///     if msg.inner.0 != "secret" {
    ///         Err(HandlerError::close_connection(
    ///             VarInt::from_u32(401),
    ///             "permission denied",
    ///         ))
    ///     } else {
    ///         msg.tx.send(()).await.ok();
    ///         Ok(())
    ///     }
    /// });
    /// ```
    ///
    /// [Error codes]: self#error-codes
    pub fn close_connection(code: VarInt, reason: impl AsRef<[u8]>) -> Self {
        Self {
            inner: HandlerErrorInner::CloseConnection {
                code,
                reason: reason.as_ref().to_vec(),
            },
        }
    }
}

/// Handles the requests that a server reads from a connection.
///
/// A server uses one handler for all its connections. Create one with:
///
/// - [`Handler::from_sender`]: sends each request to a [`LocalSender`].
/// - [`Handler::sequential`]: runs a function on each request, one at a time.
/// - [`Handler::raw`]: runs a function on the protocol enum and the streams.
///
/// A handler function can return `Err(HandlerError)` to close the connection.
/// In all other cases, handle errors internally and return `Ok(())` from
/// the handler function to read the next request from the connection.
/// A bad request also closes the connection, unless
/// [`Handler::skip_bad_requests`] is set.
///
/// Often, the protocol type cannot be inferred from the message enum,
/// so you may have to name it: `Handler::<MyProtocol>::sequential(..)`.
///
/// # Examples
///
/// ```
/// use irpc::{WithChannels, channel::oneshot, rpc::Handler, rpc_requests};
/// use serde::{Deserialize, Serialize};
///
/// #[rpc_requests(message = EchoMessage)]
/// #[derive(Debug, Serialize, Deserialize)]
/// enum EchoProtocol {
///     #[rpc(tx = oneshot::Sender<String>)]
///     #[wrap(Echo)]
///     Echo(String),
/// }
///
/// let handler: Handler<EchoProtocol> = Handler::sequential(|msg| async move {
///     match msg {
///         EchoMessage::Echo(msg) => {
///             let WithChannels { inner, tx, .. } = msg;
///             tx.send(inner.0).await.ok();
///         }
///     }
///     Ok(())
/// });
/// ```
#[derive(derive_more::Debug)]
pub struct Handler<S> {
    #[debug(skip)]
    f: HandlerFn<S>,
    skip_bad_requests: bool,
}

impl<S> Clone for Handler<S> {
    fn clone(&self) -> Self {
        Self {
            f: self.f.clone(),
            skip_bad_requests: self.skip_bad_requests,
        }
    }
}

impl<S: Service> Handler<S> {
    /// Creates a handler that runs `f` on the protocol enum and the streams of each request.
    ///
    /// The requests of a connection run one after the other, each in its span
    /// from [`Service::new_request_span`]. If `f` calls
    /// [`RemoteService::with_remote_channels`], the message gets this span.
    pub fn raw<F, Fut>(f: F) -> Self
    where
        F: Fn(S, noq::RecvStream, noq::SendStream) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = std::result::Result<(), HandlerError>> + Send + 'static,
    {
        let f: HandlerFn<S> = Arc::new(move |msg, rx, tx| Box::pin(f(msg, rx, tx)));
        Self {
            f,
            skip_bad_requests: false,
        }
    }

    /// Runs this handler on one request.
    ///
    /// Use it in a [`Handler::raw`] to pass a request to another handler.
    pub fn call(
        &self,
        request: S,
        rx: noq::RecvStream,
        tx: noq::SendStream,
    ) -> impl Future<Output = std::result::Result<(), HandlerError>> + Send + 'static {
        (self.f)(request, rx, tx)
    }

    /// Sets whether a bad request is skipped instead of closing the connection.
    ///
    /// By default, a request that is too large or does not decode closes the
    /// connection. With `true`, only the request fails, and the handler reads
    /// the next request. Use it for a protocol that adds request types over
    /// time, see [Bad requests].
    ///
    /// [Bad requests]: self#bad-requests
    pub fn skip_bad_requests(mut self, skip: bool) -> Self {
        self.skip_bad_requests = skip;
        self
    }

    /// Handles the requests of `connection` until the connection ends.
    ///
    /// Returns an error if the connection fails, or if a bad request closed it.
    pub async fn handle_connection(
        &self,
        connection: &impl IncomingRemoteConnection,
    ) -> io::Result<()> {
        loop {
            let Some((msg, carrier, rx, tx)) = self.read_next(connection).await? else {
                return Ok(());
            };
            let span = new_request_span(&msg, &rx, carrier.as_ref());
            let fut = span
                .in_scope(|| {
                    debug!("request received");
                    (self.f)(msg, rx, tx)
                })
                .instrument(span);
            if let Err(err) = fut.await {
                match err.inner {
                    HandlerErrorInner::CloseConnection { code, reason } => {
                        debug!(%code, "handler closed the connection");
                        connection.close(code, &reason);
                        return Ok(());
                    }
                }
            }
        }
    }

    /// Reads the next request, and closes the connection or skips it on a bad request.
    async fn read_next(
        &self,
        connection: &impl IncomingRemoteConnection,
    ) -> io::Result<Option<Request<S>>> {
        loop {
            match read_request_inner::<S>(connection).await {
                Ok(request) => return Ok(request),
                Err(err) => match err.error_code() {
                    None => return Err(err.into()), // A connection error.
                    Some(_) if self.skip_bad_requests => debug!("skipped bad request: {err:#}"),
                    Some(code) => {
                        connection.close(code.to_code(), err.to_string().as_bytes());
                        return Err(err.into());
                    }
                },
            }
        }
    }
}

impl<S: RemoteService> Handler<S> {
    /// Creates a handler that sends each request to `local_sender`.
    ///
    /// A full channel applies backpressure to the connection.
    pub fn from_sender(local_sender: impl Into<LocalSender<S>>) -> Self {
        let local_sender = local_sender.into();
        Self::sequential(move |msg| {
            let local_sender = local_sender.clone();
            async move {
                local_sender.send_raw(msg).await.map_err(|err| {
                    // The receiver is gone. Close as a dropped connection does.
                    warn!("handler stopped: {err:#}");
                    HandlerError::close_connection(VarInt::from_u32(0), b"")
                })
            }
        })
    }

    /// Creates a handler that runs `f` on each request, one at a time per connection.
    pub fn sequential<F, Fut>(f: F) -> Self
    where
        F: Fn(S::Message) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = std::result::Result<(), HandlerError>> + Send + 'static,
    {
        let f = Arc::new(f);
        Self::raw(move |msg, rx, tx| {
            let f = f.clone();
            async move { f(S::with_remote_channels(msg, rx, tx)).await }
        })
    }
}

impl<S: RemoteService> From<LocalSender<S>> for Handler<S> {
    fn from(local_sender: LocalSender<S>) -> Self {
        Self::from_sender(local_sender)
    }
}

/// Extension trait to [`Service`] to create a [`Service::Message`] from a [`Service`]
/// and a pair of QUIC streams.
///
/// This trait is auto-implemented when using the [`crate::rpc_requests`] macro.
pub trait RemoteService: Service + Sized {
    /// Returns the message enum for this request by combining `self` (the protocol enum)
    /// with a pair of QUIC streams for `tx` and `rx` channels.
    fn with_remote_channels(self, rx: noq::RecvStream, tx: noq::SendStream) -> Self::Message;
}

/// Abstracts over the connections that a server can read requests from.
///
/// This is implemented for noq connections, and for iroh connections with and
/// without 0-RTT. It is used by [`read_request`] and [`Handler::handle_connection`]
/// to work with all of these.
///
/// This trait is sealed: only irpc can implement it. So irpc can add methods to
/// it without a breaking change.
pub trait IncomingRemoteConnection: crate::sealed::Sealed {
    /// Accepts a single bidirectional stream.
    fn accept_bi(
        &self,
    ) -> impl Future<Output = Result<(noq::SendStream, noq::RecvStream), ConnectionError>> + Send;

    /// Closes the connection.
    fn close(&self, error_code: VarInt, reason: &[u8]);
}

/// Reads a request from a connection and converts it to a message enum.
///
/// This combines `read_request_raw` with `RemoteService::with_remote_channels`.
///
/// Returns `None` if the remote closed the connection with code 0, or if this
/// side closed it. Skips a request that the client abandoned.
///
/// Returns [`ReadRequestError::MaxSizeExceeded`],
/// [`ReadRequestError::Malformed`], or [`ReadRequestError::DecodeFailed`]
/// for a bad request. The connection is still open after these errors, see
/// [Bad requests].
///
/// [Bad requests]: self#bad-requests
pub async fn read_request<S: RemoteService>(
    connection: &impl IncomingRemoteConnection,
) -> Result<Option<S::Message>, ReadRequestError> {
    let Some((msg, carrier, rx, tx)) = read_request_inner::<S>(connection).await? else {
        return Ok(None);
    };
    let span = new_request_span(&msg, &rx, carrier.as_ref());
    Ok(Some(span.in_scope(|| {
        debug!("request received");
        S::with_remote_channels(msg, rx, tx)
    })))
}

/// Creates the span of a request, and sets its parent from the span context of the client.
fn new_request_span<S: Service>(
    msg: &S,
    recv: &noq::RecvStream,
    carrier: Option<&crate::span_propagation::SpanContextCarrier>,
) -> tracing::Span {
    let span = msg.new_request_span(recv.id().index());
    if let Some(carrier) = carrier {
        carrier.set_parent_of(&span);
    }
    span
}

/// Reads a single request from the connection.
///
/// This accepts a bi-directional stream from the connection and reads and parses the request.
///
/// Unlike [`read_request`] and a [`Handler`], this creates no request span, and
/// drops the span context of the client.
///
/// Returns `None` in the same cases as [`read_request`].
pub async fn read_request_raw<S: Service>(
    connection: &impl IncomingRemoteConnection,
) -> Result<Option<(S, noq::RecvStream, noq::SendStream)>, ReadRequestError> {
    Ok(read_request_inner::<S>(connection)
        .await?
        .map(|(msg, _carrier, rx, tx)| (msg, rx, tx)))
}

/// A request with its span context carrier and its streams.
type Request<S> = (
    S,
    Option<crate::span_propagation::SpanContextCarrier>,
    noq::RecvStream,
    noq::SendStream,
);

/// Internal: read a request and also return the propagated span context carrier.
///
/// The carrier is `Some` iff `S::SPAN_PROPAGATION` is true and the remote sent one.
async fn read_request_inner<S: Service>(
    connection: &impl IncomingRemoteConnection,
) -> Result<Option<Request<S>>, ReadRequestError> {
    loop {
        let (mut send, mut recv) = match connection.accept_bi().await {
            Ok(streams) => streams,
            Err(ConnectionError::ApplicationClosed(cause))
                if cause.error_code == VarInt::from_u32(0) =>
            {
                trace!("remote side closed connection {cause:?}");
                return Ok(None);
            }
            Err(ConnectionError::LocallyClosed) => return Ok(None),
            Err(cause) => match ConnectionClose::from_error(&cause) {
                Some(close) => return Err(e!(ReadRequestError::ConnectionClosed { close })),
                None => return Err(e!(ReadRequestError::ConnectionFailed, cause)),
            },
        };
        let err = match read_request_frame::<S>(&mut recv).await {
            ReadFrame::Request(carrier, msg) => return Ok(Some((msg, carrier, recv, send))),
            ReadFrame::MaxSizeExceeded => e!(ReadRequestError::MaxSizeExceeded),
            ReadFrame::Malformed(source) => e!(ReadRequestError::Malformed, source),
            ReadFrame::DecodeFailed(source) => e!(ReadRequestError::DecodeFailed, source),
            ReadFrame::Reset(reason) => {
                debug!(?reason, "skipped request: the client reset its stream");
                // A dropped stream finishes, which a client with an mpsc
                // response would take for the end of the response.
                send.reset(ErrorCode::RequestAborted.to_code()).ok();
                continue;
            }
            // The next `accept_bi` returns why.
            ReadFrame::ConnectionLost => continue,
        };
        if let Some(code) = err.error_code() {
            recv.stop(code.to_code()).ok();
            send.reset(code.to_code()).ok();
        }
        return Err(err);
    }
}

/// What the server reads from the stream of a request.
enum ReadFrame<S> {
    /// A complete request.
    Request(Option<crate::span_propagation::SpanContextCarrier>, S),
    /// The request is larger than [`MAX_MESSAGE_SIZE`].
    MaxSizeExceeded,
    /// The stream ended before the request was complete.
    Malformed(io::Error),
    /// The request does not decode.
    DecodeFailed(postcard::Error),
    /// The client reset the stream before the request was complete, with its code if it was not 0.
    Reset(Option<ErrorCode>),
    /// The connection was lost before the request was complete.
    ConnectionLost,
}

impl<S> ReadFrame<S> {
    /// Returns the frame for a stream that failed before the request was complete.
    fn from_read_err(err: &noq::ReadError) -> Self {
        match err {
            // Any reset, not only `ErrorCode::RequestAborted`: the update sender of a
            // request resets the stream if an update does not encode. noq drops
            // unread data on a reset, so this reset can arrive before the request.
            noq::ReadError::Reset(code) => Self::Reset(ErrorCode::from_code(*code)),
            noq::ReadError::ConnectionLost(_) => Self::ConnectionLost,
            // A server cannot get these: irpc stops a request stream only after
            // its read failed, and only a client gets `ZeroRttRejected`.
            noq::ReadError::ClosedStream | noq::ReadError::ZeroRttRejected => Self::Reset(None),
        }
    }
}

/// Reads and decodes a request.
async fn read_request_frame<S: Service>(recv: &mut noq::RecvStream) -> ReadFrame<S> {
    let size = match read_size(recv).await {
        Ok(Some(size)) => size,
        Ok(None) => {
            return ReadFrame::Malformed(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "stream ended before the request",
            ));
        }
        Err(err) => {
            return match read_error(&err) {
                Some(read_err) => ReadFrame::from_read_err(read_err),
                // The stream ended inside the size prefix.
                None => ReadFrame::Malformed(err),
            };
        }
    };
    if size > MAX_MESSAGE_SIZE {
        return ReadFrame::MaxSizeExceeded;
    }
    let mut buf = vec![0; size as usize];
    match recv.read_exact(&mut buf).await {
        Ok(()) => {}
        Err(noq::ReadExactError::ReadError(err)) => return ReadFrame::from_read_err(&err),
        Err(err @ noq::ReadExactError::FinishedEarly(_)) => {
            return ReadFrame::Malformed(io::Error::new(io::ErrorKind::UnexpectedEof, err));
        }
    }
    let decoded = if S::SPAN_PROPAGATION {
        postcard::from_bytes(&buf)
    } else {
        postcard::from_bytes(&buf).map(|msg| (None, msg))
    };
    match decoded {
        Ok((carrier, msg)) => ReadFrame::Request(carrier, msg),
        Err(err) => ReadFrame::DecodeFailed(err),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[cfg(feature = "noq_endpoint_setup")]
    #[tokio::test]
    #[n0_tracing_test::traced_test]
    async fn mpsc_frame_error_ends_receiver() -> testresult::TestResult<()> {
        use std::future::poll_fn;

        let (server, cert) = crate::noq::make_server_endpoint("127.0.0.1:0".parse()?)?;
        let client = crate::noq::make_client_endpoint("127.0.0.1:0".parse()?, &[cert.as_slice()])?;
        let connecting = client.connect(server.local_addr()?, "localhost")?;
        let accepting = async { server.accept().await.unwrap().await };
        let (client_conn, server_conn) = tokio::try_join!(connecting, accepting)?;

        for (bytes, expected_code) in [
            // An invalid bool followed by a valid bool in a separate frame.
            (vec![1, 2, 1, 0], ErrorCode::MessageDecodeFailed),
            // An overflowing length prefix.
            (vec![0xff; 10], ErrorCode::MessageMaxSizeExceeded),
        ] {
            let (mut send, _response) = client_conn.open_bi().await?;
            // Make the stream visible to the server.
            send.write_all(&[0]).await?;
            let (_response, recv) = server_conn.accept_bi().await?;
            // Seed read-ahead directly so the decode case does not depend on
            // whether the transport delivers both frames in a single read.
            let mut receiver = NoqReceiver::<bool> {
                recv,
                span: tracing::Span::none(),
                received: 0,
                finished: false,
                end: bytes.len(),
                buf: SmallVec::from_vec(bytes),
                _marker: PhantomData,
            };
            let err = poll_fn(|cx| receiver.poll_recv(cx)).await.unwrap_err();
            match expected_code {
                ErrorCode::MessageDecodeFailed => assert!(matches!(
                    err,
                    mpsc::RecvError::Rpc {
                        source: RpcRecvError::DecodeFailed { .. },
                        ..
                    }
                )),
                ErrorCode::MessageMaxSizeExceeded => assert!(matches!(
                    err,
                    mpsc::RecvError::Rpc {
                        source: RpcRecvError::MaxSizeExceeded { .. },
                        ..
                    }
                )),
                _ => unreachable!(),
            }
            for _ in 0..2 {
                assert_eq!(poll_fn(|cx| receiver.poll_recv(cx)).await?, None);
            }
            assert_eq!(send.stopped().await?, Some(expected_code.to_code()));
        }
        Ok(())
    }

    #[test]
    fn error_code_round_trips() {
        for code in ErrorCode::KNOWN {
            assert_eq!(ErrorCode::from_code(code.to_code()), Some(code));
        }
        for code in [16, 401, u64::from(u32::MAX) + 1, VarInt::MAX.into_inner()] {
            let code = VarInt::from_u64(code).unwrap();
            assert_eq!(ErrorCode::from_code(code), Some(ErrorCode::Other(code)));
        }
        assert_eq!(ErrorCode::from_code(VarInt::from_u32(0)), None);
    }

    /// Parses the next frame from `buf`, and on a whole one removes it and returns its body.
    fn take(buf: &mut Vec<u8>) -> io::Result<Result<Vec<u8>, Frame>> {
        Ok(match parse_frame(buf)? {
            Frame::Whole { body } => {
                let bytes = buf[body.clone()].to_vec();
                buf.drain(..body.end);
                Ok(bytes)
            }
            other => Err(other),
        })
    }

    fn frame(body: &[u8]) -> Vec<u8> {
        use crate::util::WriteVarintExt;
        let mut frame = Vec::new();
        frame.write_varint_u64(body.len() as u64).unwrap();
        frame.extend_from_slice(body);
        frame
    }

    #[test]
    fn parse_frame_waits_for_every_split() {
        let body = vec![7u8; 300];
        let frame = frame(&body);
        let prefix = frame.len() - body.len();
        for split in 0..frame.len() {
            let mut buf = frame[..split].to_vec();
            let frame_len = (split >= prefix).then_some(frame.len());
            assert_eq!(parse_frame(&buf).unwrap(), Frame::Partial { frame_len });
            buf.extend_from_slice(&frame[split..]);
            assert_eq!(take(&mut buf).unwrap(), Ok(body.clone()));
            assert!(buf.is_empty());
        }
    }

    #[test]
    fn parse_frame_leaves_the_next_frame() {
        let mut buf = [frame(b"one"), frame(b""), frame(b"three")].concat();
        assert_eq!(take(&mut buf).unwrap(), Ok(b"one".to_vec()));
        assert_eq!(take(&mut buf).unwrap(), Ok(b"".to_vec()));
        assert_eq!(take(&mut buf).unwrap(), Ok(b"three".to_vec()));
        assert_eq!(
            take(&mut buf).unwrap(),
            Err(Frame::Partial { frame_len: None })
        );
    }

    #[test]
    fn parse_frame_rejects_oversized_and_overlong_lengths() {
        use crate::util::WriteVarintExt;
        let mut buf = Vec::new();
        buf.write_varint_u64(MAX_MESSAGE_SIZE + 1).unwrap();
        assert_eq!(parse_frame(&buf).unwrap(), Frame::TooLarge);

        // u64::MAX is a valid varint, just too large a message.
        let max = [[0xFFu8; 9].as_slice(), &[0x01]].concat();
        assert_eq!(parse_frame(&max).unwrap(), Frame::TooLarge);

        // A 10th byte with bits beyond u64, or an 11th byte, does not fit in a
        // u64, so it is too large as well.
        let overflow = [[0xFFu8; 9].as_slice(), &[0x02]].concat();
        let overlong = [0x80u8; 10];
        for buf in [&overflow[..], &overlong[..]] {
            assert_eq!(parse_frame(buf).unwrap(), Frame::TooLarge);
        }
    }

    #[test]
    fn connection_close_shows_code_and_reason() {
        let close = |code, reason: &[u8]| {
            ConnectionClose::Remote {
                code,
                reason: reason.to_vec(),
            }
            .to_string()
        };
        let code = Some(ErrorCode::Other(VarInt::from_u32(42)));
        assert_eq!(close(None, b""), "closed by the peer");
        assert_eq!(close(None, b"bye"), "closed by the peer: bye");
        assert_eq!(close(code, b""), "closed by the peer (code 42)");
        assert_eq!(
            close(code, b"maintenance"),
            "closed by the peer (code 42): maintenance"
        );
        assert_eq!(
            close(Some(ErrorCode::RequestDecodeFailed), b""),
            "closed by the peer (request does not decode)"
        );
    }

    /// Checks that every error with a close maps it to the same io error kind, as noq does.
    #[test]
    fn connection_close_maps_to_one_io_kind() {
        let remote = ConnectionClose::Remote {
            code: None,
            reason: Vec::new(),
        };
        for (close, kind) in [
            (remote, io::ErrorKind::ConnectionAborted),
            (ConnectionClose::Local, io::ErrorKind::NotConnected),
        ] {
            let errors = [
                io::Error::from(e!(RpcSendError::ConnectionClosed {
                    close: close.clone()
                })),
                io::Error::from(e!(RpcRecvError::ConnectionClosed {
                    close: close.clone()
                })),
                io::Error::from(e!(ReadRequestError::ConnectionClosed {
                    close: close.clone()
                })),
                io::Error::from(e!(crate::RequestError::ConnectionClosed {
                    close: close.clone()
                })),
            ];
            for err in errors {
                assert_eq!(err.kind(), kind, "{err}");
            }
        }
    }

    #[test]
    fn send_error_maps_stop_codes() {
        let err = SendError::from(noq::WriteError::Stopped(
            ErrorCode::RequestDecodeFailed.to_code(),
        ));
        assert_eq!(err.remote_code(), Some(ErrorCode::RequestDecodeFailed));
        assert_eq!(io::Error::from(err).kind(), io::ErrorKind::ConnectionReset);
        let err = SendError::from(noq::WriteError::Stopped(0u32.into()));
        assert!(matches!(err, SendError::ReceiverClosed { .. }), "{err:?}");
        let err = SendError::from(noq::WriteError::Stopped(401u32.into()));
        assert_eq!(err.remote_code(), Some(ErrorCode::Other(401u32.into())));
    }

    #[test]
    fn send_error_detects_zero_rtt_rejection() {
        assert!(SendError::from(noq::WriteError::ZeroRttRejected).is_zero_rtt_rejected());
        assert!(!SendError::from(noq::WriteError::ClosedStream).is_zero_rtt_rejected());
        assert!(!SendError::from(noq::WriteError::Stopped(16u32.into())).is_zero_rtt_rejected());
    }
}