moqtap_client/draft23/

connection.rs

use std::collections::VecDeque;
use std::sync::Mutex;

use bytes::{Buf, Bytes, BytesMut};

use crate::draft23::endpoint::{Endpoint, EndpointError, FetchExistence, TrackExistence};
use crate::draft23::event::{ClientEvent, Direction, StreamKind};
use crate::draft23::fill::LocationFilter;
use crate::draft23::observer::ConnectionObserver;
use crate::draft23::session::request_id::Role;
use crate::draft23::session::setup;
use crate::malformed_tracks::MalformedTrackCondition;
use crate::track_locations::{ObjectLocation, ObjectRole, TrackObjects};
use crate::transport::{RecvStream, SendStream, Transport, TransportError};
use moqtap_codec::dispatch::{
    AnyControlMessage, AnyDatagramHeader, AnyFetchHeader, AnySubgroupHeader,
};
use moqtap_codec::draft23::data_stream::{
    FetchHeader, FetchObject, FetchObjectHeader, FetchObjectReader, GroupOrder, SubgroupObject,
    SubgroupObjectReader,
};
use moqtap_codec::draft23::error_codes::{SessionErrorCode, StreamResetErrorCode};
use moqtap_codec::draft23::message::{
    max_control_message_size, ControlMessage, MessageType, Namespace, NamespaceDone,
    PublishSkipped, RequestError, RequestOk, SubscribeOk, DEFAULT_MAX_CONTROL_MESSAGE_SIZE,
};
use moqtap_codec::error::CodecError;
use moqtap_codec::kvp::KeyValuePair;
use moqtap_codec::types::*;
use moqtap_codec::varint::VarInt;
use moqtap_codec::version::DraftVersion;

/// The ALPN identifier draft-23 uses on raw QUIC, `moqt-23`.
///
/// The version is settled by the TLS handshake before a byte of MoQT is
/// written.
///
/// This is [`DraftVersion::Draft23`]'s own
/// [`quic_alpn`](DraftVersion::quic_alpn), which is what
/// [`ClientConfig::alpn`] offers; the test below holds the two together.
pub const MOQT_ALPN: &[u8] = b"moqt-23";

/// The unidirectional stream type that marks one direction of the control
/// plane, and the SETUP message type. On draft-23 they are one number,
/// 0x2F00: Section 6.4.1 (Unidirectional Stream Types) lists it as the type of
/// a SETUP stream, and Section 9.1 gives it as the SETUP message's own type
/// field.
///
/// Because they are the same number a control stream carries no separate
/// stream header. The varint a reader uses to recognise the stream is the
/// first field of the SETUP message it then decodes, and a writer that
/// encodes a SETUP onto a fresh unidirectional stream has already written the
/// stream type by writing the message.
///
/// Encoded with MoQT's variable-length integer, whose width is the number of
/// leading 1 bits in the first byte, 0x2F00 is the two bytes `AF 00` — four
/// under RFC 9000's encoding, which this draft does not use. Read it through
/// [`DraftVersion::decode_varint`] rather than assuming a width.
pub const CONTROL_STREAM_TYPE: u64 = 0x2F00;

/// The unidirectional stream type of FETCH_HEADER, 0x05 (draft-23 Section 6.4.1).
///
/// A fill fetch stream opens with it too: draft-23 Section 3.4 has a
/// subscription carrying FILL_PARAMETERS open "a unidirectional stream
/// beginning with a FETCH_HEADER".
const FETCH_STREAM_TYPE: u64 = 0x05;

/// The application error code a request stream is reset with when the
/// requester abandons it: `CANCELLED`, 0x1.
///
/// Draft-23 has no UNSUBSCRIBE or FETCH_CANCEL. Cancelling a request is
/// resetting the bidirectional stream it was made on, and `CANCELLED` is the
/// code draft-23 Section 12.5 assigns for "The stream was cancelled by either
/// endpoint" — see [`StreamResetErrorCode::Cancelled`].
/// Taken from the codec's own registry rather than written as a literal so a
/// renumbering in a later draft cannot be missed here.
///
/// This is what [`RequestStream::cancel`] uses when no code is chosen for it,
/// and what `RequestStream`'s [`Drop`] sends.
pub const REQUEST_CANCELLED: u64 = StreamResetErrorCode::Cancelled as u64;

/// The application error code a request stream **the peer opened** is reset
/// with when this endpoint abandons it: `INTERNAL_ERROR`, 0x0.
///
/// Dropping an inbound request is not the act [`REQUEST_CANCELLED`] describes.
/// Draft-23 Section 6.4.2.3 grants a responder a cancel — "Receivers cancel
/// requests if they are unable to or choose not to respond" — but a handle
/// that fell out of scope chose nothing; it failed to serve, which is what
/// [`StreamResetErrorCode::InternalError`], Section 12.5's "An implementation
/// specific error", names. The two codes are on the wire, so a peer counting refusals
/// can tell a deliberate rejection from a dropped request only if they differ.
///
/// It is also what a responder that already answered is reset with when it is
/// dropped without finishing. A FIN there would claim the request completed,
/// and Section 6.4.2.2 says when it has not: "the publisher of an Established
/// subscription MUST send PUBLISH_DONE, before sending a FIN."
pub const REQUEST_UNANSWERED: u64 = StreamResetErrorCode::InternalError as u64;

/// Errors from the connection layer.
#[derive(Debug, thiserror::Error)]
pub enum ConnectionError {
    /// Endpoint state machine error.
    #[error("endpoint error: {0}")]
    Endpoint(#[from] EndpointError),
    /// Wire codec error.
    #[error("codec error: {0}")]
    Codec(#[from] CodecError),
    /// Transport-level error.
    #[error("transport error: {0}")]
    Transport(#[from] TransportError),
    /// Variable-length integer decoding error.
    #[error("varint error: {0}")]
    VarInt(#[from] moqtap_codec::varint::VarIntError),
    /// Control stream was not opened.
    #[error("control stream not open")]
    NoControlStream,
    /// Stream ended before a complete message was read.
    #[error("unexpected end of stream")]
    UnexpectedEnd,
    /// A data stream ended with a FIN inside an Object.
    ///
    /// draft-23 Section 11.3: "If a stream ends gracefully (i.e., the stream
    /// terminates with a FIN) in the middle of a serialized Object, the
    /// receiver SHOULD close the session with TRUNCATED_OBJECT." A FIN between
    /// two Objects is [`ConnectionError::UnexpectedEnd`].
    #[error("the data stream ended with a FIN in the middle of an Object")]
    TruncatedObject,
    /// Stream was finished by the peer.
    #[error("stream finished")]
    StreamFinished,
    /// Invalid server address string.
    #[error("invalid server address: {0}")]
    InvalidAddress(String),
    /// TLS configuration error.
    #[error("TLS config error: {0}")]
    TlsConfig(String),
    /// Data stream used out of order (e.g. object before header).
    #[error("data stream state error: {0}")]
    DataStreamState(&'static str),
    /// A control message this build decoded for draft-23 and then could not
    /// narrow to draft-23's own message type.
    ///
    /// Unreachable, and that is not the same as harmless. `read_control`
    /// decodes with this connection's own draft, so the `AnyControlMessage` it
    /// hands back can only carry this draft's variant — but the narrowing arm
    /// is compiled in every configuration anyway, under
    /// `#[allow(unreachable_patterns)]` rather than a `cfg` naming the other
    /// drafts, because such a list has to be edited in every per-draft
    /// module whenever a draft is added and a copy that omits one leaves the
    /// match non-exhaustive.
    ///
    /// Spelled as `CodecError::UnknownMessageType(0)` it would not stay inert:
    /// every draft's
    /// [`codec_session_error_code`](Connection::codec_session_error_code)
    /// answers that variant with a session close. So the day the
    /// narrowing did fail, this build's own defect would reach a caller as *the
    /// peer sent a control message type this draft does not assign, and the
    /// session must be closed with a Protocol Violation* — carrying `0x00` as
    /// the codepoint that proved it. A conformance report reading that
    /// publishes a named, well-evidenced accusation against a relay for
    /// something no relay did.
    ///
    /// A variant of its own is what stops that.
    /// [`draft_specific_cause`](Connection::draft_specific_cause) answers it
    /// [`LocalRefusal`], the facade turns that into [`ErrorCause::Facade`], and
    /// nothing downstream can read a rule out of a cause that says nothing
    /// reached the wire. What is pinned is the consequence rather than the
    /// unreachability: nothing pins the arm's reachability, which is exactly
    /// why the consequence must not be an accusation.
    ///
    /// [`LocalRefusal`]: crate::above_codec_rules::DraftSpecificCause::LocalRefusal
    /// [`ErrorCause::Facade`]: crate::dispatch::ErrorCause::Facade
    #[error(
        "a control message decoded for draft-23 did not narrow to draft-23: a defect in this build, and evidence about nothing the peer did"
    )]
    ControlMessageNarrowing,
    /// A message that begins a request stream was handed to
    /// [`Connection::send_control`]. Nothing was written.
    #[error(
        "{0:?} begins a request stream of its own and must not be written on the control stream"
    )]
    RequestOnControlStream(MessageType),
    /// A message draft-23 places on a request stream was handed to
    /// [`Connection::send_control`]. Nothing was written.
    ///
    /// Distinct from [`ConnectionError::RequestOnControlStream`], which is about
    /// a message that would *begin* a request stream of its own. This one is
    /// about a message that belongs on a request stream already open, and so has
    /// no meaning without one around it.
    #[error("{0:?} belongs on a request stream, not on the control stream")]
    RequestStreamMessageOnControlStream(MessageType),
    /// The codec refused a datagram with [`CodecError::MalformedTrack`], for a
    /// Prior Group ID Gap or Prior Object ID Gap fault (draft-23 Section 10.8,
    /// draft-23 Section 10.9), or with [`CodecError::MalformedObjectProperties`],
    /// for a property block that does not parse (draft-23 Section 10.7). Either
    /// makes the datagram's track malformed.
    ///
    /// Carries the datagram's Track Alias, which the codec's error does not
    /// name, so [`Connection::requests_to_cancel`] can name the requests to
    /// cancel. The session stays open.
    #[error("datagram for track alias {track_alias}: {error}")]
    MalformedTrackDatagram {
        /// The Track Alias the datagram named.
        track_alias: u64,
        /// The codec's refusal.
        error: CodecError,
    },
    /// A datagram carrying an Object Status arrived with bytes after its
    /// header.
    /// Draft-23 Section 11.1.1: "An Object MUST have an empty payload unless
    /// its Object Status value is registered as permitting a payload in the
    /// Object Status registry" Section 11.2.1 says the same thing about the
    /// framing: a datagram with the STATUS bit set "is present and there is no
    /// Object Payload."
    ///
    /// The codec cannot see this. `DatagramHeader::decode` stops at the end of
    /// the header and never owns the datagram's tail, so the only layer that
    /// holds both the status and the bytes after it is this one. Without the
    /// check the application is handed, say, an End-of-Group object carrying
    /// four bytes of payload — a combination the draft forbids outright.
    ///
    /// Recoverable: the drafts state the rule as a property of a conforming
    /// object, not as one of the "MUST close the session" cases, so the datagram
    /// is refused and the session left running.
    #[error(
        "datagram for object {object_id} carries {payload_len} bytes after a header \
         whose status is {status:?}, which permits no payload"
    )]
    PayloadOnStatusDatagram {
        /// Object ID from the datagram header.
        object_id: u64,
        /// How many bytes followed the header.
        payload_len: usize,
        /// The status the header declared.
        ///
        /// Spelled out in full because the glob import of `moqtap_codec::types`
        /// brings a different `ObjectStatus` into this module.
        status: Option<moqtap_codec::draft23::types::ObjectStatus>,
    },
    /// An Object arrived carrying properties on a status that is not Normal.
    ///
    /// Draft-23 Section 11.1.2: "Any Object with status Normal can have
    /// properties (Section 8.4). If an endpoint receives properties on an Object with status that
    /// is not Normal, it MUST close the session with INVALID_OBJECT_STATUS."
    ///
    /// The codec's `read_object` refuses such an Object as a bare
    /// `InvalidField`. Its `read_object_meta` frames it, which is how the
    /// receive path names the rule here and stays in step with the wire.
    #[error(
        "object {object_id} carries {properties_len} bytes of properties on status {status:?}, \
         which is not Normal"
    )]
    PropertiesOnNonNormalStatus {
        /// The Object ID the properties arrived on.
        object_id: u64,
        /// Length in bytes of the properties block.
        properties_len: usize,
        /// The Object's status, resolved through the encoding's elision rule.
        ///
        /// Spelled out in full because the glob import of `moqtap_codec::types`
        /// brings a different `ObjectStatus` into this module.
        status: moqtap_codec::draft23::types::ObjectStatus,
    },
    /// A bidirectional stream the peer opened began with a message type that
    /// does not open a request stream.
    ///
    /// Draft-23 Section 6.3: "Bidirectional streams MUST NOT begin with any
    /// other message type unless negotiated. If they do, the peer MUST close the Session with
    /// UNEXPECTED_MESSAGE." The session has already been
    /// closed on the wire by the time this is returned, and the offending
    /// stream reset.
    #[error(
        "a bidirectional stream the peer opened began with {0:?}, which does not begin a request stream; the session was closed"
    )]
    NonRequestOnRequestStream(MessageType),
    /// A `respond_*` helper was handed a request stream this endpoint opened.
    /// Nothing was written and no state moved.
    #[error(
        "this endpoint opened request {0}; only the endpoint a request stream was opened toward may answer it"
    )]
    RespondedToOwnRequest(u64),
    /// [`RequestStream::finish`] was called on a request stream the peer opened
    /// before a response had been written on it. Nothing was written.
    ///
    /// Draft-23 Section 6.4.2.2: "An endpoint MUST NOT send a FIN on a direction
    /// of a request stream until it has sent all required messages on that
    /// direction for its request type. In particular, an endpoint sending a
    /// response to a request MUST send the corresponding response message ...
    /// before sending a FIN."
    #[error(
        "request {0} has not been answered; draft-23 Section 6.4.2.2 forbids finishing a request stream before its response"
    )]
    FinBeforeResponse(u64),
    /// [`RequestStream::finish`] was called on a subscription this endpoint is
    /// the publisher of, before PUBLISH_DONE. Nothing was written.
    ///
    /// The second half of the same sentence in draft-23 Section 6.4.2.2: "the
    /// publisher of an Established subscription MUST send PUBLISH_DONE, before
    /// sending a FIN." It applies to the accepted SUBSCRIBE a responder
    /// answered with SUBSCRIBE_OK and to the PUBLISH this endpoint sent, which
    /// is why [`Connection::publish_done`] and
    /// [`Connection::publish_done_on`] are the only routes that clear it.
    #[error(
        "request {0} is an established subscription; draft-23 Section 6.4.2.2 requires PUBLISH_DONE before a FIN"
    )]
    FinBeforePublishDone(u64),
}

impl From<crate::transport::DialError> for ConnectionError {
    /// Maps a dial failure onto the variants this error already has, so a
    /// caller matches `InvalidAddress` or `TlsConfig`.
    ///
    /// A `LocalSocket` failure maps to `InvalidAddress`, which the facade reads
    /// as this side's failure
    /// ([`ErrorCause::Facade`](crate::dispatch::ErrorCause::Facade)). The dial
    /// phase stays on [`DialError::phase`](crate::transport::DialError::phase).
    fn from(e: crate::transport::DialError) -> Self {
        match e {
            // Two variants, one arm, deliberately — see above.
            crate::transport::DialError::InvalidAddress(s)
            | crate::transport::DialError::LocalSocket(s) => ConnectionError::InvalidAddress(s),
            crate::transport::DialError::TlsConfig(s) => ConnectionError::TlsConfig(s),
            crate::transport::DialError::Transport(e) => ConnectionError::Transport(e),
        }
    }
}

/// Transport type for the connection.
#[derive(Debug, Clone)]
pub enum TransportType {
    /// Raw QUIC via quinn. The `addr` field should be `host:port`.
    Quic,
    /// WebTransport via wtransport. The `url` field is the WebTransport URL.
    WebTransport {
        /// The WebTransport endpoint URL (e.g., `https://host:port/path`).
        url: String,
    },
}

/// Configuration for a MoQT client connection.
///
/// Both `draft` and `transport` are required -- there is no `Default` impl.
pub struct ClientConfig {
    /// The MoQT draft version to use (primary, determines codec/framing).
    pub draft: DraftVersion,
    /// The transport type (QUIC or WebTransport).
    pub transport: TransportType,
    /// Whether to skip TLS certificate verification (for testing).
    pub skip_cert_verification: bool,
    /// Custom CA certificates to trust (DER-encoded).
    pub ca_certs: Vec<Vec<u8>>,
    /// Setup parameters to include in CLIENT_SETUP (e.g., auth tokens).
    pub setup_parameters: Vec<KeyValuePair>,
}

impl ClientConfig {
    /// Returns the ALPN protocol identifiers for the transport.
    pub fn alpn(&self) -> Vec<Vec<u8>> {
        match &self.transport {
            TransportType::Quic => vec![self.draft.quic_alpn().to_vec()],
            TransportType::WebTransport { .. } => vec![b"h3".to_vec()],
        }
    }
}

/// A framed writer for a send stream. Handles MoQT length-prefixed framing.
pub struct FramedSendStream {
    inner: SendStream,
    draft: DraftVersion,
    /// Stateful subgroup object writer.
    subgroup_io: Option<SubgroupObjectReader>,
}

impl FramedSendStream {
    /// Create a new framed send stream for the given draft version.
    pub fn new(inner: SendStream, draft: DraftVersion) -> Self {
        Self { inner, draft, subgroup_io: None }
    }

    /// Get the transport-level stream ID.
    pub fn stream_id(&self) -> u64 {
        self.inner.stream_id()
    }

    /// Write a control message to the stream with type+length framing.
    /// Returns the raw bytes that were written (for event capture).
    ///
    /// Encodes under the default MAX_CONTROL_MESSAGE_SIZE of 65535; see
    /// [`Self::write_control_with_limit`].
    pub async fn write_control(
        &mut self,
        msg: &AnyControlMessage,
    ) -> Result<Vec<u8>, ConnectionError> {
        let mut buf = Vec::new();
        msg.encode(&mut buf)?;
        self.inner.write_all(&buf).await?;
        Ok(buf)
    }

    /// [`Self::write_control`] for a peer whose MAX_CONTROL_MESSAGE_SIZE is
    /// `limit`.
    ///
    /// draft-23 Section 9: "An endpoint MUST NOT send a control message with a
    /// length larger than the peer's MAX_CONTROL_MESSAGE_SIZE (Section
    /// 9.1.8)." A longer message is refused with
    /// [`CodecError::ControlMessageExceedsLimit`] and nothing is written.
    pub async fn write_control_with_limit(
        &mut self,
        msg: &ControlMessage,
        limit: u64,
    ) -> Result<Vec<u8>, ConnectionError> {
        let mut buf = Vec::new();
        msg.encode_with_limit(&mut buf, limit)?;
        self.inner.write_all(&buf).await?;
        Ok(buf)
    }

    /// Write a subgroup stream header. Also initializes the internal
    /// delta-encoding state used by
    /// [`FramedSendStream::write_subgroup_object`].
    ///
    /// The header is refused, and nothing is written, if its fields disagree
    /// with its own stream type. That check has to happen here rather than at
    /// the first object: the type is what every object after it is framed
    /// against, so a header that went out saying the wrong thing cannot be
    /// taken back.
    pub async fn write_subgroup_header(
        &mut self,
        header: &AnySubgroupHeader,
    ) -> Result<(), ConnectionError> {
        let mut buf = Vec::new();
        header.encode_stream_checked(&mut buf)?;
        self.inner.write_all(&buf).await?;
        // Clippy would rather see these two arms as an `if let`, and rustc rejects
        // that in a single-draft build, where the pattern is irrefutable. Only a
        // `match` satisfies both.
        #[allow(clippy::single_match)]
        match header {
            AnySubgroupHeader::Draft23(ref header) => {
                self.subgroup_io = Some(SubgroupObjectReader::new(header));
            }
            // Only this draft's header seeds the object reader. With draft 23 the only enabled
            // draft `AnySubgroupHeader` has a single variant, the arm above is exhaustive and this
            // one unreachable. Compiled in every configuration with the lint allowed, rather than
            // gated on a `cfg` naming the other drafts: such a list has to be edited in
            // every draft module whenever a draft is added, and a copy that omits one leaves this
            // match non-exhaustive.
            #[allow(unreachable_patterns)]
            _ => {}
        }
        Ok(())
    }

    /// Write a fetch response header.
    pub async fn write_fetch_header(
        &mut self,
        header: &AnyFetchHeader,
    ) -> Result<(), ConnectionError> {
        let mut buf = Vec::new();
        header.encode_stream(&mut buf);
        self.inner.write_all(&buf).await?;
        Ok(())
    }

    /// Append a draft-23 subgroup object to the stream using the
    /// stateful writer seeded from
    /// [`FramedSendStream::write_subgroup_header`].
    pub async fn write_subgroup_object(
        &mut self,
        object: &SubgroupObject,
    ) -> Result<(), ConnectionError> {
        let writer = self
            .subgroup_io
            .as_mut()
            .ok_or(ConnectionError::DataStreamState("subgroup header not written yet"))?;
        let mut buf = Vec::new();
        writer.write_object(object, &mut buf)?;
        self.inner.write_all(&buf).await?;
        Ok(())
    }

    /// Append a fetch object to the stream.
    ///
    /// The fetch stream had a header writer and no object writer, so a caller
    /// could open one and put nothing on it through this type. The subgroup
    /// stream has had both since the writer was introduced.
    ///
    /// The declared length comes from the payload rather than from the caller's
    /// field: a header that disagrees with the bytes beside it desynchronises
    /// every object after it on the stream, and nothing downstream can recover.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::Codec`] if the header's fields disagree with the
    /// Serialization Flags that announce them, which the encoder refuses rather
    /// than writing a frame its own reader cannot take apart.
    pub async fn write_fetch_object(
        &mut self,
        header: &FetchObjectHeader,
        payload: &[u8],
    ) -> Result<(), ConnectionError> {
        let mut header = header.clone();
        header.payload_length = VarInt::from_usize(payload.len());
        let mut buf = Vec::new();
        header.encode(&mut buf)?;
        buf.extend_from_slice(payload);
        self.inner.write_all(&buf).await?;
        Ok(())
    }

    /// Finish the stream (send FIN).
    pub async fn finish(&mut self) -> Result<(), ConnectionError> {
        self.inner.finish()?;
        Ok(())
    }

    /// Reset the stream with `code`, telling the peer transmission was
    /// abandoned rather than completed.
    ///
    /// Dropping a send stream sends a FIN, which claims the stream ended
    /// cleanly; this is the only way to say the opposite. See
    /// [`SendStream::reset`].
    pub fn reset(&mut self, code: u64) -> Result<(), ConnectionError> {
        self.inner.reset(code)?;
        Ok(())
    }

    /// Returns the draft version this stream is framed for.
    pub fn draft(&self) -> DraftVersion {
        self.draft
    }
}

/// What an Object Status makes of an object here.
///
/// The end-of-track status settles where the track ended and is judged
/// against nothing: draft-23 states no rule about where one may be placed. `a_track_may_end_where_it_has_already_been.rs` asserts
/// that acceptance.
///
/// Every other status is a statement about objects rather than one of them.
fn object_role(status: Option<u64>) -> ObjectRole {
    match status {
        None | Some(0x0) => ObjectRole::Produced,
        Some(0x4) => ObjectRole::EndsTrack(None),
        _ => ObjectRole::Neither,
    }
}

/// A framed reader for a recv stream. Handles MoQT varint-length decoding.
pub struct FramedRecvStream {
    inner: RecvStream,
    buf: BytesMut,
    draft: DraftVersion,
    /// Stateful subgroup object reader.
    subgroup_io: Option<SubgroupObjectReader>,
    /// Stateful fetch object reader, started by
    /// [`FramedRecvStream::begin_fetch_objects`] rather than by the fetch
    /// header, which does not carry the order.
    fetch_io: Option<FetchObjectReader>,
    /// The record this stream's objects are measured against, and the Group ID
    /// its header named.
    ///
    /// One group for the whole stream: a subgroup header names it once and no
    /// object header repeats it. `None` on a stream that was never given one -
    /// a stream for an alias no live binding names, and every stream built
    /// outside [`Connection::accept_subgroup_stream`].
    tracking: Option<(TrackObjects, TrackExistence, u64)>,
    /// What this FETCH stream's entries are judged against, when it answers a
    /// FETCH this endpoint sent.
    fetch_existence: Option<FetchExistence>,
    /// The bytes of the last control frame a capturing read refused.
    refused: Option<Vec<u8>>,
}

impl FramedRecvStream {
    /// Create a new framed receive stream for the given draft version.
    pub fn new(inner: RecvStream, draft: DraftVersion) -> Self {
        Self {
            inner,
            buf: BytesMut::with_capacity(4096),
            draft,
            subgroup_io: None,
            fetch_io: None,
            tracking: None,
            fetch_existence: None,
            refused: None,
        }
    }

    /// Get the transport-level stream ID.
    pub fn stream_id(&self) -> u64 {
        self.inner.stream_id()
    }

    /// The bytes of the control frame the last capturing read refused.
    pub(crate) fn take_refused(&mut self) -> Option<Vec<u8>> {
        self.refused.take()
    }

    /// Keep the first `len` buffered bytes as a refused frame when the read is
    /// capturing, and hand `err` back.
    fn refuse(&mut self, capture_raw: bool, len: usize, err: ConnectionError) -> ConnectionError {
        if capture_raw {
            self.refused = Some(self.buf[..len].to_vec());
        }
        err
    }

    /// Measure this stream's objects against `objects`, all of them in `group`.
    fn measure_objects_against(
        &mut self,
        objects: TrackObjects,
        existence: TrackExistence,
        group: u64,
    ) {
        self.tracking = Some((objects, existence, group));
    }

    /// Judge one object this stream carried against where its track ended.
    fn note_subgroup_object(
        &self,
        object: u64,
        status: Option<u64>,
        properties: &[u8],
    ) -> Result<(), ConnectionError> {
        let Some((objects, existence, group)) = &self.tracking else { return Ok(()) };
        let at = ObjectLocation { group: *group, object };
        let role = object_role(status);
        objects.note_past_final(at, role).map_err(|end| {
            ConnectionError::Endpoint(EndpointError::ObjectPastFinalObject {
                alias: objects.alias(),
                group: at.group,
                object: at.object,
                final_group: end.group,
                final_object: end.object,
            })
        })?;
        crate::draft23::endpoint::note_existence(existence, at, role, properties)
            .map_err(ConnectionError::Endpoint)
    }

    /// What a FIN met while an Object was being read means: a clean end when
    /// no byte of the next Object has arrived, a truncation otherwise.
    fn ended_between_objects(&self) -> ConnectionError {
        if self.buf.is_empty() {
            ConnectionError::UnexpectedEnd
        } else {
            ConnectionError::TruncatedObject
        }
    }

    /// Read more data from the stream into the internal buffer.
    async fn fill(&mut self) -> Result<bool, ConnectionError> {
        let mut tmp = [0u8; 4096];
        match self.inner.read(&mut tmp).await {
            Ok(Some(n)) => {
                self.buf.extend_from_slice(&tmp[..n]);
                Ok(true)
            }
            Ok(None) => Ok(false),
            Err(e) => Err(ConnectionError::Transport(e)),
        }
    }

    /// Ensure at least `n` bytes are available in the buffer.
    async fn ensure(&mut self, n: usize) -> Result<(), ConnectionError> {
        while self.buf.len() < n {
            if !self.fill().await? {
                return Err(ConnectionError::UnexpectedEnd);
            }
        }
        Ok(())
    }

    /// Read this stream's leading variable-length integer **without
    /// consuming it**, and return its value.
    ///
    /// Every unidirectional MoQT stream on draft-23 opens with a varint
    /// naming what it is (Section 6.4.1): 0x05 for FETCH_HEADER, 0x10-0x1D for
    /// SUBGROUP_HEADER, and [`CONTROL_STREAM_TYPE`] for SETUP. Telling the
    /// peer's control stream apart from a data stream means reading that
    /// varint, and taking it off the transport would destroy it: the control
    /// stream's type varint *is* the SETUP message's type field, so a stream
    /// whose type had been stripped would no longer decode as a SETUP.
    ///
    /// Nothing is stripped. The bytes land in this reader's own buffer, and
    /// every other method here — [`read_control`](Self::read_control),
    /// [`read_subgroup_header`](Self::read_subgroup_header),
    /// [`read_fetch_header`](Self::read_fetch_header) — decodes out of that
    /// buffer and advances it only on a successful decode. A stream this was
    /// called on is indistinguishable from one it was not, which is what
    /// makes it safe to peek a stream and then hand it to whichever reader
    /// the type turned out to call for.
    ///
    /// It reads, so it can block: a peer that opens a stream and then writes
    /// nothing leaves this pending until a byte arrives or the stream ends.
    ///
    /// # Errors
    ///
    /// - [`ConnectionError::UnexpectedEnd`] if the stream ends before a whole
    ///   varint has arrived.
    /// - [`ConnectionError::Transport`] if the peer reset the stream.
    /// - [`ConnectionError::VarInt`] if the bytes are not a valid varint.
    ///
    /// Whatever did arrive stays in the buffer in every case.
    pub async fn peek_stream_type(&mut self) -> Result<u64, ConnectionError> {
        self.ensure(1).await?;
        let type_len = self.draft.varint_len(self.buf[0]);
        self.ensure(type_len).await?;
        let mut cursor = &self.buf[..type_len];
        Ok(self.draft.decode_varint(&mut cursor)?.into_inner())
    }

    /// Read a control message from the stream.
    ///
    /// When `capture_raw` is true, the returned tuple includes a clone of the
    /// framed wire bytes (for observer emission). When false, the second
    /// element is `None` and the payload clone is skipped.
    pub async fn read_control(
        &mut self,
        capture_raw: bool,
    ) -> Result<(AnyControlMessage, Option<Vec<u8>>), ConnectionError> {
        self.read_control_with_limit(capture_raw, DEFAULT_MAX_CONTROL_MESSAGE_SIZE).await
    }

    /// [`read_control`](Self::read_control) under the MAX_CONTROL_MESSAGE_SIZE
    /// this endpoint advertised.
    ///
    /// Draft-23 Section 9: "If an endpoint receives a control message with a
    /// length larger than its MAX_CONTROL_MESSAGE_SIZE, it MUST close the
    /// session with FIELD_LENGTH_EXCEEDED." The Length is checked before the
    /// body is buffered, so an oversized message costs no memory.
    pub async fn read_control_with_limit(
        &mut self,
        capture_raw: bool,
        limit: u64,
    ) -> Result<(AnyControlMessage, Option<Vec<u8>>), ConnectionError> {
        self.refused = None;
        // Read type ID varint
        self.ensure(1).await?;
        let type_len = self.draft.varint_len(self.buf[0]);
        self.ensure(type_len).await?;

        let mut cursor = &self.buf[..type_len];
        if let Err(e) = self.draft.decode_varint(&mut cursor) {
            return Err(self.refuse(capture_raw, type_len, e.into()));
        }

        // Draft-23 Section 9: "Message Length (vi64)".
        let (payload_len, len_field_size) = if self.draft.uses_fixed_length_framing() {
            self.ensure(type_len + 2).await?;
            let hi = self.buf[type_len] as usize;
            let lo = self.buf[type_len + 1] as usize;
            ((hi << 8) | lo, 2)
        } else {
            self.ensure(type_len + 1).await?;
            let payload_len_start = type_len;
            let payload_len_varint_len = self.draft.varint_len(self.buf[payload_len_start]);
            self.ensure(type_len + payload_len_varint_len).await?;
            let header_len = type_len + payload_len_varint_len;
            let mut cursor = &self.buf[payload_len_start..header_len];
            let length = match self.draft.decode_varint(&mut cursor) {
                Ok(length) => length.into_inner(),
                Err(e) => return Err(self.refuse(capture_raw, header_len, e.into())),
            };
            if length > limit {
                // Only the type and the length are buffered, and they are what
                // the refusal is about.
                let err = ConnectionError::Codec(CodecError::ControlMessageExceedsLimit {
                    length,
                    limit,
                });
                return Err(self.refuse(capture_raw, header_len, err));
            }
            (length as usize, payload_len_varint_len)
        };

        // Read full payload
        let total = type_len + len_field_size + payload_len;
        self.ensure(total).await?;

        // Capture raw bytes only if requested (observer attached).
        let raw = capture_raw.then(|| self.buf[..total].to_vec());

        // Now decode the whole message
        let mut frame = &self.buf[..total];
        let decoded = if self.draft == DraftVersion::Draft23 {
            ControlMessage::decode_with_limit(&mut frame, limit).map(AnyControlMessage::Draft23)
        } else {
            AnyControlMessage::decode(self.draft, &mut frame)
        };
        let msg = match decoded {
            Ok(msg) => msg,
            Err(e) => return Err(self.refuse(capture_raw, total, e.into())),
        };
        self.buf.advance(total);
        Ok((msg, raw))
    }

    /// Read a subgroup stream header. Also initializes the internal
    /// delta-decoding state.
    pub async fn read_subgroup_header(&mut self) -> Result<AnySubgroupHeader, ConnectionError> {
        self.ensure(1).await?;
        loop {
            let mut cursor = &self.buf[..];
            match AnySubgroupHeader::decode(self.draft, &mut cursor) {
                Ok(header) => {
                    let consumed = self.buf.len() - cursor.remaining();
                    self.buf.advance(consumed);
                    // Clippy would rather see these two arms as an `if let`, and rustc rejects
                    // that in a single-draft build, where the pattern is irrefutable. Only a
                    // `match` satisfies both.
                    #[allow(clippy::single_match)]
                    match header {
                        AnySubgroupHeader::Draft23(ref header) => {
                            self.subgroup_io = Some(SubgroupObjectReader::new(header));
                        }
                        // Only this draft's header seeds the object reader. With draft 23 the only
                        // enabled draft `AnySubgroupHeader` has a single variant, the arm above is
                        // exhaustive and this one unreachable. Compiled in every configuration with
                        // the lint allowed, rather than gated on a `cfg` naming every other
                        // draft: such a list has to be edited in every draft module whenever a
                        // draft is added, and a copy that omits one leaves this match
                        // non-exhaustive.
                        #[allow(unreachable_patterns)]
                        _ => {}
                    }
                    return Ok(header);
                }
                Err(e) if e.is_incomplete() => {
                    if !self.fill().await? {
                        return Err(ConnectionError::UnexpectedEnd);
                    }
                }
                Err(e) => return Err(ConnectionError::Codec(e)),
            }
        }
    }

    /// Read a fetch response header.
    pub async fn read_fetch_header(&mut self) -> Result<AnyFetchHeader, ConnectionError> {
        self.ensure(1).await?;
        loop {
            let mut cursor = &self.buf[..];
            match AnyFetchHeader::decode(self.draft, &mut cursor) {
                Ok(header) => {
                    let consumed = self.buf.len() - cursor.remaining();
                    self.buf.advance(consumed);
                    return Ok(header);
                }
                Err(e) if e.is_incomplete() => {
                    if !self.fill().await? {
                        return Err(ConnectionError::UnexpectedEnd);
                    }
                }
                Err(e) => return Err(ConnectionError::Codec(e)),
            }
        }
    }

    /// Read the next draft-23 subgroup object from this stream using
    /// the stateful reader seeded by
    /// [`FramedRecvStream::read_subgroup_header`].
    ///
    /// Errors with [`ConnectionError::PropertiesOnNonNormalStatus`] on an
    /// Object that carries properties on a status other than Normal, which
    /// draft-23 Section 11.1.2 answers with a session close. The object is
    /// consumed from the stream before the check, so the reader stays in step
    /// with the wire and a caller that reports the violation and reads on sees
    /// the following object rather than a re-parse of this one.
    pub async fn read_subgroup_object(&mut self) -> Result<SubgroupObject, ConnectionError> {
        if self.subgroup_io.is_none() {
            return Err(ConnectionError::DataStreamState("subgroup header not read yet"));
        }
        loop {
            let reader = self.subgroup_io.as_mut().unwrap();
            let mut probe = reader.clone();
            let mut cursor = &self.buf[..];
            match probe.read_object(&mut cursor) {
                Ok(obj) => {
                    let consumed = self.buf.len() - cursor.remaining();
                    self.buf.advance(consumed);
                    *reader = probe;
                    if !obj.properties_permitted() {
                        return Err(ConnectionError::PropertiesOnNonNormalStatus {
                            object_id: obj.object_id.into_inner(),
                            properties_len: obj.extension_headers.len(),
                            status: obj.status(),
                        });
                    }
                    self.note_subgroup_object(
                        obj.object_id.into_inner(),
                        obj.object_status.map(|s| s as u64),
                        &obj.extension_headers,
                    )?;
                    return Ok(obj);
                }
                Err(e) if e.is_incomplete() => {
                    if !self.fill().await? {
                        return Err(self.ended_between_objects());
                    }
                }
                // The codec refuses properties on a non-Normal status as
                // `InvalidField`, one of many readings of that variant. Its
                // framing-only reader accepts them, which tells this case apart
                // from the others and keeps the stream in step.
                Err(CodecError::InvalidField) => {
                    let mut framing = reader.clone();
                    let mut cursor = &self.buf[..];
                    if let Ok(meta) = framing.read_object_meta(&mut cursor) {
                        let status = meta
                            .status
                            .and_then(moqtap_codec::draft23::types::ObjectStatus::from_u64);
                        if let Some(status) = status.filter(|s| {
                            meta.extension_headers_len > 0
                                && *s != moqtap_codec::draft23::types::ObjectStatus::Normal
                        }) {
                            let consumed = self.buf.len() - cursor.remaining();
                            self.buf.advance(consumed);
                            *reader = framing;
                            return Err(ConnectionError::PropertiesOnNonNormalStatus {
                                object_id: meta.object_id,
                                properties_len: meta.extension_headers_len as usize,
                                status,
                            });
                        }
                    }
                    return Err(ConnectionError::Codec(CodecError::InvalidField));
                }
                Err(e) => return Err(ConnectionError::Codec(e)),
            }
        }
    }

    /// Read the next draft-23 fetch header from this stream.
    pub async fn read_fetch_stream_header(&mut self) -> Result<FetchHeader, ConnectionError> {
        loop {
            let mut cursor = &self.buf[..];
            match FetchHeader::decode(&mut cursor) {
                Ok(hdr) => {
                    let consumed = self.buf.len() - cursor.remaining();
                    self.buf.advance(consumed);
                    return Ok(hdr);
                }
                Err(e) if e.is_incomplete() => {
                    if !self.fill().await? {
                        return Err(ConnectionError::UnexpectedEnd);
                    }
                }
                Err(e) => return Err(ConnectionError::Codec(e)),
            }
        }
    }

    /// Stop accepting data on this stream with `code` as the `STOP_SENDING`
    /// application error code, discarding anything unread.
    ///
    /// Dropping a receive stream also stops it, but with a hard-coded 0. See
    /// [`RecvStream::stop`].
    pub fn stop(&mut self, code: u64) -> Result<(), ConnectionError> {
        self.inner.stop(code)?;
        Ok(())
    }

    /// Wait for the peer to reset this stream, consuming nothing.
    ///
    /// See [`RecvStream::received_reset`] for what `Ok(None)` means and why a
    /// caller must not re-poll after it.
    pub async fn received_reset(&mut self) -> Result<Option<u64>, ConnectionError> {
        Ok(self.inner.received_reset().await?)
    }

    /// Start reading fetch objects on this stream, in a response whose Groups
    /// arrive in `group_order`.
    ///
    /// Draft-23 has to be told. Section 11.4.1.1 makes a
    /// Group ID Delta count downward under Descending and upward under
    /// Ascending, so the same bytes are two different Locations and nothing on
    /// the data stream says which. The order is the one the **request** asked
    /// for — Section 3.2.1: "A publisher MUST send fetched groups in the
    /// requested group order (see Section 9.19.8)." — carried by the
    /// GROUP_ORDER parameter on the FETCH, or by its absence, which Section
    /// 9.19.8 reads as Ascending. Either way it is a control message this
    /// stream never sees.
    ///
    /// Call it after [`FramedRecvStream::read_fetch_header`] and before the
    /// first [`FramedRecvStream::read_fetch_object`]; calling it again restarts
    /// the running state, which is what a second fetch stream on the same
    /// connection would want and what the middle of one would not.
    pub fn begin_fetch_objects(&mut self, group_order: GroupOrder) {
        self.fetch_io = Some(FetchObjectReader::new(group_order));
        if let Some(existence) = &mut self.fetch_existence {
            existence.begin(group_order);
        }
    }

    /// Read the next draft-23 fetch object and its payload.
    ///
    /// The mirror of [`FramedSendStream::write_fetch_object`]. What comes back
    /// is a [`FetchObject`] rather than a header: draft-23's reader resolves the
    /// delta-encoded fields, so the Group ID, Subgroup ID, Object ID and
    /// Priority it carries are the frame's own values rather than differences
    /// from the frame before. `object.header` is the frame exactly as it
    /// arrived, for a caller forwarding the bytes on.
    ///
    /// The payload comes back with it for the reason the codec leaves it on the
    /// wire: `payload_length` says how many bytes follow, and a reader that
    /// takes the wrong number of them desynchronises every later object on the
    /// stream. Doing it here is the only place that count and the buffer are
    /// both in hand.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::DataStreamState`] when
    /// [`FramedRecvStream::begin_fetch_objects`] has not been called,
    /// [`ConnectionError::UnexpectedEnd`] when the stream ends inside the header
    /// or inside the payload it declared, and [`ConnectionError::Codec`] on
    /// every rule Section 11.4.1.1 states — a first Object that inherits, a
    /// delta that runs off either end of the space, or an Object ID above
    /// 2^64-1.
    ///
    /// A Prior Group ID Gap or Prior Object ID Gap fault the stream shows is
    /// returned as [`CodecError::MalformedTrack`], and a property block that
    /// does not parse as [`CodecError::MalformedObjectProperties`];
    /// [`Connection::requests_to_cancel_for_fetch_stream`] names the requests to
    /// cancel.
    pub async fn read_fetch_object(&mut self) -> Result<(FetchObject, Vec<u8>), ConnectionError> {
        if self.fetch_io.is_none() {
            return Err(ConnectionError::DataStreamState("fetch object reader not started"));
        }
        let object = loop {
            let reader = self.fetch_io.as_mut().unwrap();
            // Advanced on a probe and committed only once the whole header was
            // there to read: a reader advanced by a short read would resolve
            // the next Object against a half-read one.
            let mut probe = reader.clone();
            let mut cursor = &self.buf[..];
            match probe.read_object_header(&mut cursor) {
                Ok(object) => {
                    let consumed = self.buf.len() - cursor.remaining();
                    self.buf.advance(consumed);
                    *reader = probe;
                    break object;
                }
                Err(e) if e.is_incomplete() => {
                    if !self.fill().await? {
                        return Err(self.ended_between_objects());
                    }
                }
                Err(e) => return Err(ConnectionError::Codec(e)),
            }
        };
        let payload = self.read_object_payload(&object.header.payload_length).await?;
        if let Some(existence) = &mut self.fetch_existence {
            existence.note_entry(&object).map_err(ConnectionError::Endpoint)?;
        }
        Ok((object, payload))
    }

    /// Take the `length` payload bytes that follow a fetch object's header.
    ///
    /// Separate from the header read because the header is decoded from a probe
    /// cursor that may have to be retried after a fill, and the payload is a
    /// flat byte count that never is.
    async fn read_object_payload(&mut self, length: &VarInt) -> Result<Vec<u8>, ConnectionError> {
        let length = length.into_inner() as usize;
        self.ensure(length).await.map_err(|e| match e {
            ConnectionError::UnexpectedEnd => ConnectionError::TruncatedObject,
            other => other,
        })?;
        let payload = self.buf[..length].to_vec();
        self.buf.advance(length);
        Ok(payload)
    }

    /// Returns the draft version this stream is framed for.
    pub fn draft(&self) -> DraftVersion {
        self.draft
    }
}

/// Which of the seven message types draft-23 Section 6.3 lets a bidirectional
/// stream begin with opened a request stream.
///
/// Draft-23 Section 6.3: "A request stream begins with one of these seven
/// message types: TRACK_STATUS, SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE,
/// SUBSCRIBE_NAMESPACE, and SUBSCRIBE_TRACKS. Bidirectional streams MUST NOT
/// begin with any other message type unless negotiated."
///
/// The set is per draft. A
/// seven-variant enum here is what makes it impossible to label a draft-23
/// request stream with a kind this draft does not have.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestKind {
    /// TRACK_STATUS, 0x0D.
    TrackStatus,
    /// SUBSCRIBE, 0x03.
    Subscribe,
    /// PUBLISH, 0x1D.
    Publish,
    /// FETCH, 0x16. Draft-23 has one shape of FETCH: Section 9.11 has no `Fetch Type` field.
    Fetch,
    /// PUBLISH_NAMESPACE, 0x06.
    PublishNamespace,
    /// SUBSCRIBE_NAMESPACE, 0x50.
    SubscribeNamespace,
    /// SUBSCRIBE_TRACKS, 0x51.
    SubscribeTracks,
}

impl RequestKind {
    /// The message type a request stream of this kind begins with.
    pub const fn message_type(self) -> MessageType {
        match self {
            RequestKind::TrackStatus => MessageType::TrackStatus,
            RequestKind::Subscribe => MessageType::Subscribe,
            RequestKind::Publish => MessageType::Publish,
            RequestKind::Fetch => MessageType::Fetch,
            RequestKind::PublishNamespace => MessageType::PublishNamespace,
            RequestKind::SubscribeNamespace => MessageType::SubscribeNamespace,
            RequestKind::SubscribeTracks => MessageType::SubscribeTracks,
        }
    }

    /// The kind of request stream `ty` opens, or `None` when it opens none.
    ///
    /// The inverse of [`message_type`](Self::message_type) and the classifier
    /// the accept path runs on the first message of a bidirectional stream the
    /// peer opened. `None` is the UNEXPECTED_MESSAGE case of draft-23 Section 6.3.
    ///
    /// Like [`starts_a_request_stream`] the match is exhaustive over
    /// [`MessageType`] with **no wildcard arm**, so a message type added in a
    /// later draft stops this compiling until someone classifies it; the unit
    /// test below holds the two functions to the same answer for every
    /// assigned type, so neither can drift from the other.
    pub const fn from_message_type(ty: MessageType) -> Option<RequestKind> {
        match ty {
            MessageType::TrackStatus => Some(RequestKind::TrackStatus),
            MessageType::Subscribe => Some(RequestKind::Subscribe),
            MessageType::Publish => Some(RequestKind::Publish),
            MessageType::Fetch => Some(RequestKind::Fetch),
            MessageType::PublishNamespace => Some(RequestKind::PublishNamespace),
            MessageType::SubscribeNamespace => Some(RequestKind::SubscribeNamespace),
            MessageType::SubscribeTracks => Some(RequestKind::SubscribeTracks),
            MessageType::Setup
            | MessageType::GoAway
            | MessageType::Namespace
            | MessageType::NamespaceDone
            | MessageType::PublishSkipped
            | MessageType::RequestUpdate
            | MessageType::SubscribeOk
            | MessageType::RequestOk
            | MessageType::RequestError
            | MessageType::PublishDone
            // // Section 9.10 puts it on "a subscription's
            // bidirectional stream", which is a stream a SUBSCRIBE or a PUBLISH
            // already opened, so it begins none of its own.
            | MessageType::PublishStateNotify => None,
        }
    }
}

/// Which side opened the bidirectional stream a request travels on.
///
/// Draft-23 Section 6.3 gives every request a bidirectional stream, and either
/// endpoint may open one. The two directions are not symmetric — one side owes
/// a response and the other is waiting for it — so a [`RequestStream`] carries
/// this to say which side of that it is on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequestOrigin {
    /// This endpoint opened the stream and wrote the request on it. What comes
    /// back is a response, and dropping the handle cancels the request.
    Local,
    /// The peer opened the stream; this endpoint owes it a response. What
    /// comes back is a follow-up to the peer's request, never a response, and
    /// dropping the handle abandons a request that was asked of us.
    Peer,
}
/// Whether draft-23 Table 5 places this message on a request stream that is
/// already open.
///
/// All five of the messages Table 5 marks "Request" without their beginning one:
/// REQUEST_UPDATE modifies the request its stream carries (Section 9.5),
/// NAMESPACE and NAMESPACE_DONE report namespaces on the SUBSCRIBE_NAMESPACE
/// stream that asked for them (Section 9.15 and Section 9.16), PUBLISH_SKIPPED
/// names a track on the SUBSCRIBE_TRACKS stream that asked for it (Section
/// 9.17), and PUBLISH_STATE_NOTIFY reports a change on the
/// subscription's own stream (Section 9.10).
///
/// `Endpoint::receive_message` already closes the session over all five when
/// they arrive on the control stream. Without this the client would write on the
/// control stream exactly what its own peer half refuses to read there.
fn belongs_on_a_request_stream(ty: MessageType) -> bool {
    matches!(
        ty,
        MessageType::RequestUpdate
            | MessageType::Namespace
            | MessageType::NamespaceDone
            | MessageType::PublishSkipped
            | MessageType::PublishStateNotify
    )
}

/// Whether `ty` is one of the seven message types draft-23 Section 6.3 lets a
/// bidirectional stream begin with.
///
/// The match is exhaustive over [`MessageType`] and deliberately has **no
/// wildcard arm**. That is the drift guard: `MessageType` is not
/// `#[non_exhaustive]`, so the day a draft gains a message type this stops
/// compiling until someone says here whether the new type opens a request
/// stream. A wildcard would silently answer "no" for it.
///
/// Classification is over the typed `MessageType`, never over a raw `u64`,
/// because the number alone does not say which registry it came from: on
/// draft-23, 0x50 is SUBSCRIBE_NAMESPACE as a control message type and also a
/// SUBGROUP_HEADER as a unidirectional stream type.
pub const fn starts_a_request_stream(ty: MessageType) -> bool {
    match ty {
        // The seven that begin a request stream.
        MessageType::TrackStatus
        | MessageType::Subscribe
        | MessageType::Publish
        | MessageType::Fetch
        | MessageType::PublishNamespace
        | MessageType::SubscribeNamespace
        | MessageType::SubscribeTracks => true,
        // Messages that cannot begin a request stream. SETUP is the control
        // stream's own type varint. NAMESPACE, NAMESPACE_DONE, PUBLISH_SKIPPED
        // and REQUEST_UPDATE all travel ON a request stream — Table 5 gives
        // every one of them the Stream value "Request" — but none of them
        // opens one: the first three report on the namespace or publication of
        // the stream that is already there, and REQUEST_UPDATE modifies a
        // request that already has a stream, draft-23 Section 9.5 putting it
        // "on the same bidi stream as the request". This function answers
        // *does this message open a stream*, not *does it belong on one*.
        //
        // GOAWAY is here because it opens nothing, not because it is confined
        // to the control stream: it has no Request ID, so it has one wire form for both places. It
        // may follow a
        // request on an already-open request stream — see
        // [`Connection::send_on_request_stream`], which refuses nothing.
        MessageType::Setup
        | MessageType::GoAway
        | MessageType::Namespace
        | MessageType::NamespaceDone
        | MessageType::PublishSkipped
        | MessageType::RequestUpdate => false,
        // Responses. They cannot begin a stream: they arrive on the request
        // stream their request opened, which is why they carry no request id
        // of their own on this draft. PUBLISH_OK is a REQUEST_OK (Section 1.5).
        MessageType::SubscribeOk
        | MessageType::RequestOk
        | MessageType::RequestError
        | MessageType::PublishDone => false,
        // Not a response — Section 9.10 makes it unilateral, answered with
        // nothing — but it opens no stream either: it travels on the
        // subscription's, which a SUBSCRIBE or a PUBLISH already opened.
        MessageType::PublishStateNotify => false,
    }
}

/// One request and its answer, on a bidirectional stream of their own.
///
/// Draft-23 Section 6.3 keeps requests off the control plane: each request is
/// the first message on a bidirectional stream it opens, and the response
/// comes back on that same stream. Responses carry no request id on this
/// draft — **the stream is the correlation**, which is why this handle exists
/// and why a bare request id is no longer enough to find an answer.
///
/// # Reading and writing go through the connection
///
/// This handle owns both halves of the stream but not the session, so the
/// endpoint state machine and the observer stay where they were. Read a
/// response with [`Connection::recv_on_request_stream`], write a follow-up with
/// [`Connection::send_on_request_stream`], and cancel with
/// [`Connection::cancel_request_stream`].
///
/// [`cancel`](Self::cancel) and [`peer_cancelled`](Self::peer_cancelled) are on
/// the handle because they touch the stream and nothing else, and [`Drop`]
/// needs the first of them. Neither moves the endpoint's record of the request,
/// which is why the connection carries a pair of its own.
///
/// # Dropping this cancels the request
///
/// A dropped handle resets the send half and sends `STOP_SENDING` on the
/// receive half, both with [`REQUEST_CANCELLED`], unless the stream was
/// already cancelled or finished. Letting the default drop stand would send a
/// FIN instead, telling the peer the request ended *cleanly* when it was
/// abandoned.
///
/// The consequence is sharp and worth stating: a live subscription's request
/// stream must be **held for the subscription's life**, because PUBLISH_DONE
/// arrives on it. Keeping only [`request_id`](Self::request_id) and letting
/// the handle fall out of scope cancels the subscription.
///
/// What a drop cannot do is say so at the endpoint. [`Drop`] holds the stream
/// and not the session, so the request stays where it was in the endpoint's
/// record while the stream it travelled on is gone. Call
/// [`Connection::cancel_request_stream`] wherever that record matters.
///
/// # Which side opened it changes what this handle does
///
/// [`origin`](Self::origin) says whether this endpoint opened the stream or
/// accepted it, and three behaviours turn on it: reads dispatch as responses
/// or as follow-ups to the peer's request, the `respond_*` helpers refuse a
/// stream this endpoint opened, and [`Drop`] resets with
/// [`REQUEST_UNANSWERED`] rather than [`REQUEST_CANCELLED`]. Everything else —
/// [`cancel`](Self::cancel), [`peer_cancelled`](Self::peer_cancelled),
/// [`Connection::send_on_request_stream`] — is the same in both directions.
/// Draft-23 Section 6.4.2.3 is explicit that a cancel is available to both:
/// "Senders cancel requests if the response is no longer of interest;
/// Receivers cancel requests if they are unable to or choose not to respond."
///
/// [`finish`](Self::finish) is the one act that is constrained rather than
/// merely different: Section 6.4.2.2 forbids a FIN before the messages that
/// direction still owes.
///
/// All fields are private so the shape can grow without breaking callers.
#[must_use = "dropping a request stream cancels the request; hold it until the response arrives"]
pub struct RequestStream {
    send: FramedSendStream,
    recv: FramedRecvStream,
    request_id: VarInt,
    kind: RequestKind,
    /// The unidirectional stream this request's objects are being served on.
    ///
    /// Only a FETCH has one, and only once the caller has opened it through
    /// [`Connection::open_fetch_stream_on`]. Held here rather than in a table
    /// on the connection because the request stream is already the thing that
    /// knows what this request still owes, and because a handle kept beside
    /// the request cannot outlive it.
    fetch_data: Option<FramedSendStream>,
    draft: DraftVersion,
    stream_id: u64,
    cancelled: bool,
    finished: bool,
    origin: RequestOrigin,
    /// Whether a `respond_*` helper has written a response on this stream.
    /// True on a [`RequestOrigin::Peer`] stream from the response written on
    /// it, and on a [`RequestOrigin::Local`] one from the answer to an update
    /// the peer sent, which is the only response this endpoint writes on a
    /// stream of its own.
    responded: bool,
    /// Whether this endpoint is the publisher of a subscription established on
    /// this stream and has not yet sent PUBLISH_DONE.
    ///
    /// True from the PUBLISH this endpoint sent, and from the SUBSCRIBE_OK it
    /// wrote answering a peer's SUBSCRIBE; false again once PUBLISH_DONE is
    /// out. It is what draft-23 Section 6.4.2.2's second clause is checked
    /// against in [`finish`](Self::finish).
    owes_publish_done: bool,
}

impl RequestStream {
    /// The request id the endpoint allocated for this request.
    ///
    /// Useful for logging and for endpoint calls that still take one. It is
    /// not enough to find the response: draft-23 responses carry no request
    /// id, so only this stream identifies them.
    pub fn request_id(&self) -> VarInt {
        self.request_id
    }

    /// Which of the seven request types opened this stream.
    pub fn kind(&self) -> RequestKind {
        self.kind
    }

    /// The transport-level stream identifier, the same one
    /// [`ClientEvent::StreamOpened`] reports for data streams.
    pub fn stream_id(&self) -> u64 {
        self.stream_id
    }

    /// The draft version this stream is framed for.
    pub fn draft(&self) -> DraftVersion {
        self.draft
    }

    /// Which side opened this stream.
    ///
    /// [`RequestOrigin::Peer`] means this endpoint owes a response and the
    /// `respond_*` helpers apply; [`RequestOrigin::Local`] means it is waiting
    /// for one.
    pub fn origin(&self) -> RequestOrigin {
        self.origin
    }

    /// Whether a response has been written on this stream by one of the
    /// `respond_*` helpers.
    ///
    /// On a [`RequestOrigin::Local`] stream this says an update the peer sent
    /// was answered here, not that the request itself was: that one is
    /// answered by the peer.
    pub fn responded(&self) -> bool {
        self.responded
    }

    /// Whether this endpoint still owes the peer a PUBLISH_DONE on this stream.
    ///
    /// True for a PUBLISH this endpoint sent and for a peer's SUBSCRIBE it
    /// answered with SUBSCRIBE_OK — the two ways draft-23 Section 6.4.2.2's
    /// "publisher of an Established subscription" is reached — until
    /// [`Connection::publish_done`] or [`Connection::publish_done_on`] has run.
    /// While it is true, [`finish`](Self::finish) refuses.
    pub fn owes_publish_done(&self) -> bool {
        self.owes_publish_done
    }

    /// The stream this request's objects are being served on, if one is open.
    ///
    /// Only a FETCH answered through
    /// [`Connection::open_fetch_stream_on`](Connection::open_fetch_stream_on)
    /// has one. Writing objects goes through this rather than through a handle
    /// the caller keeps, so that the connection can still reach the stream
    /// when a rule says to reset it.
    pub fn fetch_data(&mut self) -> Option<&mut FramedSendStream> {
        self.fetch_data.as_mut()
    }

    /// Reset the fetch data stream, if one was opened, and forget it.
    ///
    /// The sentence that requires this names no error code, so the code comes
    /// from the registry rather than from here: CANCELLED is "the stream was
    /// cancelled by either endpoint", which is what a publisher abandoning the
    /// objects it was serving has done.
    fn reset_fetch_data(&mut self) {
        if let Some(mut framed) = self.fetch_data.take() {
            let _ = framed.reset(StreamResetErrorCode::Cancelled as u64);
        }
    }

    /// Whether [`cancel`](Self::cancel) has already run on this handle.
    ///
    /// Says nothing about the peer: a peer reset is learned from
    /// [`peer_cancelled`](Self::peer_cancelled) or from the next read.
    pub fn is_cancelled(&self) -> bool {
        self.cancelled
    }

    /// Cancel the request by resetting the stream, handing the peer `code`.
    ///
    /// Draft-23 has no UNSUBSCRIBE or FETCH_CANCEL: resetting the request stream is how a request
    /// is withdrawn. Both halves are shut — a QUIC bidirectional stream has two
    /// independent halves, so resetting only the send half would leave the
    /// peer free to keep writing a response nobody will read. The send half is
    /// reset with `code` and the receive half is stopped with the same value.
    ///
    /// [`REQUEST_CANCELLED`] is the ordinary choice. Draft-23 Section 12.5
    /// says an application SHOULD take the code from the Stream Reset Error
    /// Codes registry when resetting, or sending STOP_SENDING on, any stream —
    /// request streams included — so [`StreamResetErrorCode`] is where a value
    /// other than the default should come from. The parameter is a plain `u64`
    /// because the registry reserves greasing code points that have no variant;
    /// a caller with a named code has [`StreamResetErrorCode::as_u64`].
    ///
    /// **This is the stream and nothing else.** The endpoint's record of the
    /// request does not move, so a response already in flight is still accepted
    /// after this returns. [`Connection::cancel_request_stream`] does both and
    /// is what a caller holding a connection should reach for; this stays
    /// because [`Drop`] has no connection to reach.
    ///
    /// Idempotent, and it retires the [`Drop`] behaviour: a cancelled handle
    /// does nothing further when it goes out of scope. Errors from a stream
    /// that was already reset or stopped are swallowed for the same reason —
    /// the request is cancelled either way.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::Transport`] carrying [`TransportError::Write`] if
    /// `code` is outside the QUIC varint range (`0..2^62`). Nothing is sent
    /// in that case, and the handle is *not* marked cancelled, so a caller
    /// can retry with a representable code.
    pub fn cancel(&mut self, code: u64) -> Result<(), ConnectionError> {
        if self.cancelled {
            return Ok(());
        }
        // Reject an unrepresentable code before either half is touched, so a
        // failed call leaves the stream exactly as it was.
        if code > MAX_QUIC_VARINT {
            return Err(ConnectionError::Transport(TransportError::Write(format!(
                "error code {code} exceeds the varint range"
            ))));
        }
        self.cancelled = true;
        // Already-finished or already-reset halves report StreamClosed; the
        // request ends up cancelled regardless, so neither is worth raising.
        let _ = self.send.reset(code);
        let _ = self.recv.stop(code);
        Ok(())
    }

    /// Wait for the peer to cancel this request, consuming nothing.
    ///
    /// A caller applying backpressure is deliberately not calling
    /// [`Connection::recv_on_request_stream`], which is the only other place a
    /// peer reset surfaces — so without this the abandonment goes unobserved
    /// for as long as the backpressure lasts. This grants no flow-control
    /// credit and is cancel-safe.
    ///
    /// Returns `Ok(Some(code))` with the peer's application error code, or
    /// `Ok(None)` meaning **no reset is observable, now or ever — stop
    /// asking**. A caller that re-polls after `Ok(None)` spins.
    ///
    /// Like [`cancel`](Self::cancel), this records nothing at the endpoint.
    /// [`Connection::peer_cancelled_on_request_stream`] is the same wait with
    /// the record attached.
    ///
    /// On WebTransport this always answers `Ok(None)`: `wtransport` exposes no
    /// reset-only observable, so a WebTransport caller learns of a peer cancel
    /// on its next read and not before.
    pub async fn peer_cancelled(&mut self) -> Result<Option<u64>, ConnectionError> {
        self.recv.received_reset().await
    }

    /// Finish the send half cleanly, leaving the receive half open.
    ///
    /// Draft-23 Section 6.4.2.2 settles what a FIN means, which earlier drafts
    /// left open: "A FIN only indicates that an endpoint will send no further
    /// messages in that direction; it is not a request cancellation." A
    /// requester may therefore FIN before its response arrives — "A requester,
    /// with the exception of the sender of PUBLISH, MAY FIN immediately after
    /// sending a message if it will not send a REQUEST_UPDATE" — and the
    /// receive half stays open to carry the response, which is why this touches
    /// only the send half.
    ///
    /// What the same section forbids is finishing early: "An endpoint MUST NOT
    /// send a FIN on a direction of a request stream until it has sent all
    /// required messages on that direction for its request type", and "An
    /// endpoint that receives a FIN before all required messages have arrived
    /// treats the request as failed." The section names the two messages that
    /// are always required — "an endpoint sending a response to a request MUST
    /// send the corresponding response message, and the publisher of an
    /// Established subscription MUST send PUBLISH_DONE" — and both are things
    /// this handle knows about itself, so both are checked here and **nothing
    /// is written** when either is outstanding.
    ///
    /// A responder that owes a response has not sent one:
    /// [`ConnectionError::FinBeforeResponse`]. A publisher that owes a
    /// PUBLISH_DONE, whether from a PUBLISH it sent or a peer's SUBSCRIBE it
    /// accepted, gets [`ConnectionError::FinBeforePublishDone`] — which is why
    /// [`Connection::publish_done`] and [`Connection::publish_done_on`] are the
    /// routes that end a publication, rather than a bare `finish`.
    ///
    /// What is not checked is everything the caller alone knows: a
    /// SUBSCRIBE_NAMESPACE responder deciding it has no more namespaces to
    /// announce, or a requester deciding it will send no REQUEST_UPDATE. The
    /// section leaves those to the endpoint, and so does this.
    ///
    /// A finished handle, like a cancelled one, does nothing further on
    /// [`Drop`]. That matters here: the default drop resets both halves, which
    /// Section 6.4.2.2 contrasts with a FIN as the abrupt close, so a request
    /// that ended cleanly must come through this to avoid being reported as
    /// cancelled.
    pub async fn finish(&mut self) -> Result<(), ConnectionError> {
        if self.finished || self.cancelled {
            return Ok(());
        }
        if self.origin == RequestOrigin::Peer && !self.responded {
            return Err(ConnectionError::FinBeforeResponse(self.request_id.into_inner()));
        }
        if self.owes_publish_done {
            return Err(ConnectionError::FinBeforePublishDone(self.request_id.into_inner()));
        }
        self.finished = true;
        self.send.finish().await
    }
}

impl Drop for RequestStream {
    /// Reset the request unless it was already cancelled or finished.
    ///
    /// See the type-level note: the default drop would FIN the send half,
    /// which claims a clean end for a request the caller walked away from —
    /// and on draft-23 a FIN is a stronger claim than on earlier drafts, since
    /// Section 6.4.2.2 has the receiver of one *treat the request as failed*
    /// only when messages are still owed and complete otherwise.
    ///
    /// The code says which walking away it was. A stream this endpoint opened
    /// is cancelled — [`REQUEST_CANCELLED`] — which is the requester act
    /// Section 6.4.2.3 describes. A stream the peer opened is reset with
    /// [`REQUEST_UNANSWERED`] whether or not a response was already written:
    /// before one, the request was never served; after one, the obligations
    /// that follow it are still outstanding.
    fn drop(&mut self) {
        if self.cancelled || self.finished {
            return;
        }
        let code = match self.origin {
            RequestOrigin::Local => REQUEST_CANCELLED,
            RequestOrigin::Peer => REQUEST_UNANSWERED,
        };
        let _ = self.send.reset(code);
        let _ = self.recv.stop(code);
    }
}

/// Holds a peer-opened stream pair while its first message is being read, and
/// puts it back on the connection's queue if that read is abandoned.
///
/// [`Connection::accept_request_stream`] awaits a whole control message, and a
/// caller may drop that future — a `select!` against a shutdown signal is the
/// ordinary reason. Without this the stream, and every byte already read off
/// it into the reader's buffer, would go with the future: the peer would see a
/// request stream reset for no reason it could act on.
///
/// [`Drop`] is the only place this can run, because a cancelled future is
/// never polled again. Every path that finishes — success or error — takes the
/// pair out first, so a pair still present when this drops was cancelled.
struct PendingInbound<'a> {
    pair: Option<(FramedSendStream, FramedRecvStream)>,
    queue: &'a Mutex<VecDeque<(FramedSendStream, FramedRecvStream)>>,
}

impl Drop for PendingInbound<'_> {
    fn drop(&mut self) {
        if let Some(pair) = self.pair.take() {
            // Front, not back: this stream arrived before anything still
            // queued behind it, and a partially read message must not be
            // handed out after a stream that arrived later.
            self.queue.lock().unwrap_or_else(|p| p.into_inner()).push_front(pair);
        }
    }
}

/// The largest value a QUIC application error code can carry, `2^62 - 1`.
///
/// Checked by [`RequestStream::cancel`] before either half of the stream is
/// touched, so an unrepresentable code cannot half-cancel a request.
const MAX_QUIC_VARINT: u64 = (1u64 << 62) - 1;

/// A live MoQT connection over QUIC or WebTransport, combining the endpoint
/// state machine with actual network I/O.
pub struct Connection {
    transport: Transport,
    endpoint: Endpoint,
    draft: DraftVersion,
    /// The MAX_CONTROL_MESSAGE_SIZE this endpoint put in its SETUP, which
    /// bounds every control message it reads (draft-23 Section 9.1.8).
    recv_control_limit: u64,
    /// The MAX_CONTROL_MESSAGE_SIZE the server's SETUP advertised, which every
    /// control message after this endpoint's SETUP is encoded under.
    send_control_limit: u64,
    control_send: Option<FramedSendStream>,
    control_recv: Option<FramedRecvStream>,
    observer: Option<Box<dyn ConnectionObserver>>,
    /// Setup events buffered during `connect()` and replayed when an
    /// observer attaches via `set_observer` — without this, an observer
    /// attached after `connect` returns would never see the handshake.
    pending_events: Vec<ClientEvent>,
    /// The server's half of the setup handshake, kept whole.
    ///
    /// The endpoint acts on the parameters it recognises and retains none of
    /// them, and which parameters a server sends — in what order, with what
    /// values — is the sharpest thing a session says about the implementation
    /// behind it.
    server_setup: AnyControlMessage,
    /// The framed wire bytes of [`Self::server_setup`].
    server_setup_raw: Option<Vec<u8>>,
    /// Unidirectional streams accepted while `connect` was looking for the
    /// peer's control stream, in arrival order, and then any stream that is
    /// not a FETCH_HEADER one that a fetch or fill accept met.
    ///
    /// Data streams are allowed to arrive before the control streams on this
    /// draft, so the search cannot assume the first unidirectional stream is
    /// the control one — and dropping the ones that are not would silently
    /// lose objects the peer already sent.
    /// [`accept_subgroup_stream`](Connection::accept_subgroup_stream) empties
    /// this before it accepts anything new.
    ///
    /// Behind a mutex because that method takes `&self`. The lock is only
    /// ever held for a push or a pop, never across an await.
    deferred_uni: Mutex<VecDeque<FramedRecvStream>>,
    /// FETCH_HEADER streams, fetch and fill alike, accepted by a call that
    /// wanted another kind, in arrival order. The type varint is still on
    /// the front of each.
    /// [`accept_fetch_stream`](Connection::accept_fetch_stream) and
    /// [`accept_fill_stream`](Connection::accept_fill_stream) empty this
    /// before they accept anything new.
    deferred_fetch: Mutex<VecDeque<FramedRecvStream>>,
    /// Bidirectional streams the peer opened that
    /// [`accept_request_stream`](Connection::accept_request_stream) took off
    /// the transport but did not finish reading a first message from, because
    /// its future was dropped. In arrival order.
    ///
    /// Without this a caller could not put `accept_request_stream` in a
    /// `select!` at all: losing the race would lose a stream the peer had
    /// already opened and, with it, whatever of the request had arrived.
    /// [`accept_request_stream`](Connection::accept_request_stream) empties
    /// this before it accepts anything new.
    ///
    /// Behind a mutex for the same reason `deferred_uni` is: the lock is only
    /// ever held for a push or a pop, never across an await.
    pending_inbound: Mutex<VecDeque<(FramedSendStream, FramedRecvStream)>>,
}

impl Connection {
    /// Connect to a MoQT server as a client.
    ///
    /// Establishes a QUIC or WebTransport connection (based on
    /// `config.transport`), brings up the control plane, performs the SETUP
    /// handshake, and returns a ready-to-use connection.
    ///
    /// # The control plane is a pair of unidirectional streams
    ///
    /// Draft-23 Section 6.3: "MOQT uses a pair of unidirectional streams for
    /// creating the session and exchanging control messages. Each peer opens
    /// one control stream beginning with a SETUP message." So each direction
    /// is a separate stream opened by the peer that writes on it. This opens
    /// one with `open_uni` and writes SETUP on it, then finds the peer's by
    /// accepting unidirectional streams until one leads with
    /// [`CONTROL_STREAM_TYPE`].
    ///
    /// Nothing is written ahead of the SETUP: 0x2F00 is both the SETUP
    /// message type and the unidirectional stream type for a control stream,
    /// so the message's own first field is the stream header. See
    /// [`CONTROL_STREAM_TYPE`].
    ///
    /// A bidirectional stream is *not* the control stream here — the same
    /// section makes it a request stream, one that begins with TRACK_STATUS,
    /// SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE or
    /// SUBSCRIBE_TRACKS: "Bidirectional streams MUST NOT begin with any other
    /// message type unless negotiated. If they do, the peer MUST close the Session with
    /// UNEXPECTED_MESSAGE." A SETUP written on a bidirectional
    /// stream is exactly that case, so a peer that enforces the topology
    /// answers it by closing the session.
    ///
    /// # Unidirectional streams that arrive before the peer's control stream
    ///
    /// They are kept, not dropped. Section 6.3 expects them: "Unidirectional
    /// streams containing Objects or bidirectional stream(s) beginning with a
    /// request message could arrive prior to the control streams, in which
    /// case the data SHOULD be buffered until both control streams arrive and
    /// setup is complete." Each such stream is set aside and handed to
    /// [`accept_subgroup_stream`](Self::accept_subgroup_stream) in arrival
    /// order, ahead of any newly accepted stream. Only the leading type
    /// varint is read from them here; the rest stays on the transport, unread
    /// and still flow-controlled, so nothing is buffered in this process
    /// beyond those few bytes.
    ///
    /// One limit worth knowing: the search waits for each stream's type
    /// varint in turn, so a peer that opens a unidirectional stream and then
    /// writes nothing on it stalls the handshake behind that stream.
    pub async fn connect(addr: &str, config: ClientConfig) -> Result<Self, ConnectionError> {
        // PATH is for native QUIC only, and the transport is known here and
        // nowhere further in. Refusing before dialling means a session that
        // the server would close on sight is never opened.
        setup::validate_client_path_transport(
            &config.setup_parameters,
            matches!(config.transport, TransportType::WebTransport { .. }),
        )
        .map_err(EndpointError::from)?;

        let transport = match &config.transport {
            TransportType::Quic => Self::connect_quic(addr, &config).await?,
            TransportType::WebTransport { url } => {
                let url = url.clone();
                Self::connect_webtransport(&url, &config).await?
            }
        };

        Self::adopt(transport, config).await
    }

    /// Run the MoQT setup handshake over a transport somebody else established.
    ///
    /// For choosing the draft from what the server selected: dial once through
    /// [`crate::transport::dial_quic`] offering every ALPN, then bring the
    /// connection to the module its answer names. [`Self::connect`] cannot do
    /// this — it derives its single ALPN from the draft it was given.
    ///
    /// `config.draft` must match this module. The transport is adopted as
    /// given; nothing here re-checks the ALPN it was negotiated with.
    pub async fn adopt(
        transport: Transport,
        config: ClientConfig,
    ) -> Result<Self, ConnectionError> {
        Self::adopt_with(transport, config, None).await
    }

    /// [`Self::adopt`] with `observer` attached before the handshake.
    ///
    /// The SETUP this side sends is reported as it is written, and a SETUP from
    /// the peer that this side refuses as [`ClientEvent::ControlMessageRefused`].
    /// [`Self::set_observer`] after `adopt` sees the same handshake events, but
    /// only once the handshake has succeeded.
    pub async fn adopt_observed(
        transport: Transport,
        config: ClientConfig,
        observer: Box<dyn ConnectionObserver>,
    ) -> Result<Self, ConnectionError> {
        Self::adopt_with(transport, config, Some(observer)).await
    }

    /// [`Self::adopt`], reporting to `observer` from the first byte written.
    pub(crate) async fn adopt_with(
        transport: Transport,
        config: ClientConfig,
        observer: Option<Box<dyn ConnectionObserver>>,
    ) -> Result<Self, ConnectionError> {
        let draft = config.draft;
        // PATH is for native QUIC only, and the transport is known here and
        // nowhere further in. Refusing before dialling means a session that
        // the server would close on sight is never opened.
        setup::validate_client_path_transport(
            &config.setup_parameters,
            matches!(config.transport, TransportType::WebTransport { .. }),
        )
        .map_err(EndpointError::from)?;

        // Send half of the control plane: one unidirectional stream whose
        // first message is SETUP, which is also its stream header.
        let send = transport.open_uni().await?;
        let mut control_send = FramedSendStream::new(send, draft);

        // Perform setup handshake (draft-23: no versions)
        let mut endpoint = Endpoint::new(Role::Client);
        endpoint.connect()?;
        let setup_msg = endpoint.send_setup(config.setup_parameters.clone())?;
        // Section 9.1.8: "Until an endpoint receives the peer's SETUP, it uses
        // the default value, including for SETUP itself." That binds what the
        // peer sends; what this endpoint accepts is what it advertised.
        let recv_control_limit = max_control_message_size(&config.setup_parameters)?;
        let any_setup = AnyControlMessage::Draft23(setup_msg);
        let raw_setup = control_send.write_control(&any_setup).await?;
        // Reported as written, so a peer that never answers still leaves it
        // observed.
        if let Some(obs) = &observer {
            obs.on_event_owned(ClientEvent::ControlMessage {
                direction: Direction::Send,
                message: any_setup.clone(),
                stream_id: None,
                raw: Some(raw_setup.clone()),
            });
        }

        // Receive half: the peer's control stream is whichever unidirectional
        // stream leads with CONTROL_STREAM_TYPE.
        let mut deferred_uni: VecDeque<FramedRecvStream> = VecDeque::new();
        let mut deferred_fetch: VecDeque<FramedRecvStream> = VecDeque::new();
        let mut control_recv = loop {
            let recv = transport.accept_uni().await?;
            let mut framed = FramedRecvStream::new(recv, draft);
            match framed.peek_stream_type().await {
                Ok(CONTROL_STREAM_TYPE) => break framed,
                Ok(FETCH_STREAM_TYPE) => deferred_fetch.push_back(framed),
                // Every other type is a data stream — and so is a stream that
                // ended or failed before its type arrived, not because it is
                // one but because there is nothing left to decide with. The
                // data path sees the same end one read later and reports it
                // the way it reports every other. Treating it as the control
                // stream would hand the session's control plane to a stream
                // that carried nothing.
                _ => deferred_uni.push_back(framed),
            }
        };

        let (server_setup, raw_server_setup) =
            match control_recv.read_control_with_limit(true, recv_control_limit).await {
                Ok(read) => read,
                Err(e) => {
                    if let ConnectionError::Codec(codec) = &e {
                        if let Some(code) = Self::codec_session_error_code(codec) {
                            let wire_code = u32::try_from(code.as_u64()).unwrap_or(u32::MAX);
                            transport.close(wire_code, codec.to_string().as_bytes());
                        }
                    }
                    if let Some(obs) = &observer {
                        if let Some(event) =
                            Self::refused_event(&e, None, control_recv.take_refused())
                        {
                            obs.on_event_owned(event);
                        }
                    }
                    return Err(e);
                }
            };
        // Unified SETUP in draft-23: server responds with the same message type.
        //
        // Section 9: "An endpoint MUST NOT send a control message with a
        // length larger than the peer's MAX_CONTROL_MESSAGE_SIZE (Section
        // 9.1.8)." The decoder has already refused a value that does not parse.
        let send_control_limit = match &server_setup {
            AnyControlMessage::Draft23(ControlMessage::Setup(ref s)) => {
                if let Err(e) = endpoint.receive_setup(s) {
                    let err = ConnectionError::from(e);
                    Self::refuse_setup(observer.as_deref(), &err, &server_setup, &raw_server_setup);
                    return Err(err);
                }
                match s.max_control_message_size() {
                    Ok(limit) => limit,
                    Err(e) => {
                        let err = ConnectionError::Codec(e);
                        Self::refuse_setup(
                            observer.as_deref(),
                            &err,
                            &server_setup,
                            &raw_server_setup,
                        );
                        return Err(err);
                    }
                }
            }
            _ => {
                return Err(ConnectionError::Endpoint(EndpointError::NotActive));
            }
        };

        let mut pending_events = vec![
            ClientEvent::ControlMessage {
                direction: Direction::Send,
                message: any_setup,
                stream_id: None,
                raw: Some(raw_setup),
            },
            ClientEvent::ControlMessage {
                direction: Direction::Receive,
                message: server_setup.clone(),
                stream_id: None,
                raw: raw_server_setup.clone(),
            },
            ClientEvent::SetupComplete { negotiated_version: 0xff000000 + 23 },
        ];

        if let Some(obs) = &observer {
            // The SETUP this side sent was reported as it was written.
            for event in pending_events.drain(..).skip(1) {
                obs.on_event_owned(event);
            }
        }

        Ok(Self {
            transport,
            endpoint,
            draft,
            recv_control_limit,
            send_control_limit,
            control_send: Some(control_send),
            control_recv: Some(control_recv),
            observer,
            pending_events,
            server_setup,
            server_setup_raw: raw_server_setup,
            deferred_uni: Mutex::new(deferred_uni),
            deferred_fetch: Mutex::new(deferred_fetch),
            pending_inbound: Mutex::new(VecDeque::new()),
        })
    }

    /// Establish a raw QUIC connection.
    ///
    /// Offers this draft's ALPN alone; [`crate::transport::dial_quic`] holds the
    /// TLS and endpoint setup.
    async fn connect_quic(addr: &str, config: &ClientConfig) -> Result<Transport, ConnectionError> {
        let (transport, _negotiated) = crate::transport::dial_quic(
            addr,
            &crate::transport::QuicDialOptions {
                skip_cert_verification: config.skip_cert_verification,
                ca_certs: config.ca_certs.clone(),
                ..crate::transport::QuicDialOptions::new(config.alpn())
            },
        )
        .await?;
        Ok(transport)
    }

    /// Establish a WebTransport connection.
    ///
    /// [`crate::transport::dial_webtransport`] holds the TLS and endpoint
    /// setup, exactly as `connect_quic` above defers its own. That is not
    /// only deduplication: both dials must trust the same roots. Settling trust
    /// at this call site instead — from `wtransport`'s own builder settings, or
    /// from a second config of this draft's own — puts the decision in two
    /// places, where it can stop matching what the QUIC dial trusts, so one
    /// relay would pass on one transport and fail on the other and a caller's
    /// private CA would reach only the dials whose call site installed it.
    /// Both ask the same function what to trust.
    #[cfg(feature = "webtransport")]
    async fn connect_webtransport(
        url: &str,
        config: &ClientConfig,
    ) -> Result<Transport, ConnectionError> {
        Ok(crate::transport::dial_webtransport(
            url,
            &crate::transport::QuicDialOptions {
                skip_cert_verification: config.skip_cert_verification,
                ca_certs: config.ca_certs.clone(),
                // The draft's own protocol identifier, which this draft
                // negotiates in `WT-Available-Protocols` rather than in
                // CLIENT_SETUP: "The client includes MOQT protocol identifiers
                // in the WT-Available-Protocols header". `config.alpn()` above
                // is `h3`, which is the HTTP/3 name and settles no version.
                wt_protocols: vec![config.draft.quic_alpn().to_vec()],
                ..crate::transport::QuicDialOptions::new(config.alpn())
            },
        )
        .await?)
    }

    /// Stub for when the webtransport feature is not enabled.
    #[cfg(not(feature = "webtransport"))]
    async fn connect_webtransport(
        _url: &str,
        _config: &ClientConfig,
    ) -> Result<Transport, ConnectionError> {
        Err(ConnectionError::Transport(TransportError::Connect(
            "webtransport feature not enabled".into(),
        )))
    }

    // -- Observer ---------------------------------------------------

    /// Attach an observer. Buffered handshake events from `connect()` are
    /// flushed in arrival order before this returns.
    pub fn set_observer(&mut self, observer: Box<dyn ConnectionObserver>) {
        self.observer = Some(observer);
        for event in self.pending_events.drain(..) {
            if let Some(ref obs) = self.observer {
                obs.on_event_owned(event);
            }
        }
    }

    /// Remove the observer.
    pub fn clear_observer(&mut self) {
        self.observer = None;
    }

    /// How the QUIC connection was closed, once it has been. See
    /// [`TransportClose`](crate::transport::TransportClose).
    pub fn transport_close(&self) -> Option<crate::transport::TransportClose> {
        self.transport.transport_close()
    }

    /// Emit an event to the observer, if one is attached.
    fn emit(&self, event: ClientEvent) {
        if let Some(ref obs) = self.observer {
            obs.on_event_owned(event);
        }
    }

    /// Report a received frame this endpoint refused, when an observer is
    /// attached and the refusal is the peer's: see [`Self::refused_event`].
    fn emit_refused(&self, err: &ConnectionError, stream_id: Option<u64>, raw: Option<Vec<u8>>) {
        if self.observer.is_some() {
            if let Some(event) = Self::refused_event(err, stream_id, raw) {
                self.emit(event);
            }
        }
    }

    /// The [`ClientEvent::ControlMessageRefused`] for `raw`, when `err` refuses
    /// what the peer sent: the decoder refused it, or it broke a rule
    /// [`EndpointError::fault`] or [`Self::draft_specific_cause`] lays on the
    /// peer.
    fn refused_event(
        err: &ConnectionError,
        stream_id: Option<u64>,
        raw: Option<Vec<u8>>,
    ) -> Option<ClientEvent> {
        use crate::above_codec_rules::{DraftSpecificCause, EndpointFault};
        let peers = match err {
            ConnectionError::Codec(_) | ConnectionError::VarInt(_) => true,
            ConnectionError::Endpoint(e) => matches!(e.fault(), EndpointFault::Peer(_)),
            other => matches!(
                Self::draft_specific_cause(other),
                Some(DraftSpecificCause::PeerViolation { .. } | DraftSpecificCause::Codec { .. })
            ),
        };
        let raw = raw.filter(|_| peers)?;
        Some(ClientEvent::ControlMessageRefused { stream_id, raw })
    }

    /// Report a peer's SETUP the handshake refused to an observer attached
    /// before it: the message as received, then the refusal.
    fn refuse_setup(
        observer: Option<&dyn ConnectionObserver>,
        err: &ConnectionError,
        setup: &AnyControlMessage,
        raw: &Option<Vec<u8>>,
    ) {
        let Some(obs) = observer else { return };
        obs.on_event_owned(ClientEvent::ControlMessage {
            direction: Direction::Receive,
            message: setup.clone(),
            stream_id: None,
            raw: raw.clone(),
        });
        if let Some(event) = Self::refused_event(err, None, raw.clone()) {
            obs.on_event_owned(event);
        }
    }

    // -- Control message I/O ----------------------------------------

    /// Send a control message on the control stream.
    ///
    /// Wraps the draft-23 message in `AnyControlMessage::Draft23` for framing.
    /// Of the messages draft-23 Table 5 does not place on a request stream,
    /// exactly two reach the wire: SETUP, written by
    /// [`connect`](Self::connect) as the control stream's own type varint, and
    /// GOAWAY, which is the one message the table gives the Stream value
    /// "Control, Request" and which belongs here when it drains the session.
    ///
    /// # Four messages that this refuses
    ///
    /// Table 5 gives NAMESPACE (0x8), NAMESPACE_DONE (0xE), PUBLISH_SKIPPED
    /// (0xF) and REQUEST_UPDATE (0x2) the Stream value **Request**, not
    /// Control. REQUEST_UPDATE is sent "on the same bidi stream as the request"
    /// it modifies (Section 9.5); the other three report on the namespace or
    /// the publication of the request stream they arrive on. A conforming peer
    /// answers any of the four on the control stream by closing the session,
    /// and so does this client's own [`Endpoint::receive_message`] — writing
    /// one here produces a frame this implementation would itself refuse.
    ///
    /// All four have somewhere better to go, and none of them should come
    /// through here. REQUEST_UPDATE goes on the stream its request opened, with
    /// [`send_on_request_stream`](Self::send_on_request_stream). NAMESPACE and
    /// NAMESPACE_DONE go on a peer's SUBSCRIBE_NAMESPACE stream, with
    /// [`namespace_on`](Self::namespace_on) and
    /// [`namespace_done_on`](Self::namespace_done_on); PUBLISH_SKIPPED goes on
    /// a peer's SUBSCRIBE_TRACKS stream, with
    /// [`publish_skipped_on`](Self::publish_skipped_on). This method refuses
    /// all four with [`ConnectionError::RequestStreamMessageOnControlStream`]
    /// and writes nothing.
    ///
    /// [`Endpoint::receive_message`]: crate::draft23::endpoint::Endpoint::receive_message
    ///
    /// GOAWAY is correct here and is not confined here. It has no Request ID, so its control-stream
    /// and request-stream forms are one wire form;
    /// [`send_on_request_stream`](Self::send_on_request_stream)
    /// will put it on an open request stream.
    ///
    /// # Requests are refused here
    ///
    /// Draft-23 Section 6.3 keeps requests off the control plane: "In addition
    /// to the control streams, this specification uses bidirectional streams
    /// to carry requests. A request stream begins with one of these seven
    /// message types: TRACK_STATUS, SUBSCRIBE, PUBLISH, FETCH,
    /// PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE, and SUBSCRIBE_TRACKS." The
    /// response comes back on that same bidirectional stream, and resetting it
    /// cancels the request.
    ///
    /// Handing one of those seven to this method returns
    /// [`ConnectionError::RequestOnControlStream`] and writes **nothing** —
    /// an enforcing peer sees no bytes at all, not a misplaced request. Use
    /// the typed helpers, which open a bidirectional stream each:
    /// [`subscribe`](Self::subscribe), [`fetch`](Self::fetch),
    /// [`publish`](Self::publish), [`track_status`](Self::track_status),
    /// [`publish_namespace`](Self::publish_namespace),
    /// [`subscribe_namespace`](Self::subscribe_namespace) and
    /// [`subscribe_tracks`](Self::subscribe_tracks).
    ///
    /// Response types are still permitted here, and need not be. A
    /// response written here will be refused by a conforming peer, whose
    /// endpoint answers a response on the control stream with an error; the
    /// route that is correct is
    /// [`accept_request_stream`](Self::accept_request_stream) and the
    /// `respond_*` helpers, which write on the stream the request arrived on
    /// and take the Request ID off the handle rather than from the caller.
    /// [`publish_done`](Self::publish_done) and
    /// [`publish_done_on`](Self::publish_done_on) do not come through here
    /// either: each takes the request stream its publication travels on.
    pub async fn send_control(&mut self, msg: &ControlMessage) -> Result<(), ConnectionError> {
        let ty = msg.message_type();
        if starts_a_request_stream(ty) {
            return Err(ConnectionError::RequestOnControlStream(ty));
        }
        // What this endpoint refuses to receive on the control stream it must
        // not write there either, or the client emits frames its own peer half
        // would close the session over.
        if belongs_on_a_request_stream(ty) {
            return Err(ConnectionError::RequestStreamMessageOnControlStream(ty));
        }
        let any = AnyControlMessage::Draft23(msg.clone());
        let send = self.control_send.as_mut().ok_or(ConnectionError::NoControlStream)?;
        let raw = send.write_control_with_limit(msg, self.send_control_limit).await?;
        self.emit(ClientEvent::ControlMessage {
            direction: Direction::Send,
            message: any,
            stream_id: None,
            raw: Some(raw),
        });
        Ok(())
    }

    /// Read the next control message from the control stream.
    ///
    /// Returns the `AnyControlMessage` and also extracts the draft-23
    /// `ControlMessage` for internal endpoint dispatch.
    pub async fn recv_control(&mut self) -> Result<ControlMessage, ConnectionError> {
        self.recv_control_raw().await.map(|(msg, _)| msg)
    }

    /// [`Self::recv_control`], keeping the frame's wire bytes when an observer
    /// is attached.
    async fn recv_control_raw(
        &mut self,
    ) -> Result<(ControlMessage, Option<Vec<u8>>), ConnectionError> {
        let recv = self.control_recv.as_mut().ok_or(ConnectionError::NoControlStream)?;
        let capture_raw = self.observer.is_some();
        let read = recv.read_control_with_limit(capture_raw, self.recv_control_limit).await;
        let (any, raw) = match read {
            Ok(v) => v,
            // Section 6.3: "A control stream MUST NOT be closed at the
            // underlying transport layer during the session's lifetime. If the
            // control stream is closed, the peer MUST close the session with
            // CLOSED_CONTROL_STREAM."
            Err(
                e @ (ConnectionError::StreamFinished
                | ConnectionError::UnexpectedEnd
                | ConnectionError::Transport(TransportError::StreamReset(_))),
            ) => {
                let code = SessionErrorCode::ClosedControlStream.as_u64();
                self.close(u32::try_from(code).unwrap_or(u32::MAX), b"control stream closed");
                return Err(e);
            }
            Err(e) => {
                let raw = recv.take_refused();
                let err = self.close_for_codec(e);
                self.emit_refused(&err, None, raw);
                return Err(err);
            }
        };
        if capture_raw {
            self.emit(ClientEvent::ControlMessage {
                direction: Direction::Receive,
                message: any.clone(),
                stream_id: None,
                raw: raw.clone(),
            });
        }
        // Unwrap to draft-23 for the endpoint
        match any {
            AnyControlMessage::Draft23(msg) => Ok((msg, raw)),
            // `AnyControlMessage` carries one variant per enabled draft feature. With draft 23 the
            // only one enabled the arm above is exhaustive and this rejection arm unreachable.
            // Compiled in every configuration with the lint allowed, rather than gated on a `cfg`
            // naming the other drafts: such a list has to be edited in every draft module
            // whenever a draft is added, and a copy that omits one leaves this match
            // non-exhaustive.
            #[allow(unreachable_patterns)]
            _ => Err(ConnectionError::ControlMessageNarrowing),
        }
    }

    /// Read and dispatch the next incoming control message through the
    /// endpoint state machine. Returns the decoded message for inspection.
    ///
    /// Responses never arrive here. Draft-23 responses carry no request id
    /// and belong on the request stream that asked for them, so the endpoint
    /// refuses a response that turns up on the control stream. Read them with
    /// [`recv_on_request_stream`](Self::recv_on_request_stream).
    ///
    /// A violation the draft answers with a session close is closed on the
    /// wire here, before the error is returned: the QUIC connection is closed
    /// with the code [`EndpointError::session_error_code`] names, so the peer
    /// that broke the rule learns of it rather than only the local endpoint.
    pub async fn recv_and_dispatch(&mut self) -> Result<ControlMessage, ConnectionError> {
        let (msg, raw) = self.recv_control_raw().await?;
        if let Err(e) = self.endpoint.receive_message(msg.clone()) {
            let err = self.close_if_session_fatal(e);
            self.emit_refused(&err, None, raw);
            return Err(err);
        }

        // Emit draining event if this was a GoAway
        if let ControlMessage::GoAway(ref ga) = msg {
            self.emit(ClientEvent::Draining { new_session_uri: ga.new_session_uri.clone() });
        }

        Ok(msg)
    }

    // -- Request streams --------------------------------------------

    /// Open the bidirectional stream a request will be carried on.
    ///
    /// Opened *before* the endpoint allocates a request id, so a transport
    /// that refuses a new stream — the peer's `initial_max_streams_bidi` is
    /// exhausted, the connection is gone — costs nothing. The endpoint has no
    /// way to abandon a request it has already allocated, so every failure
    /// that can be moved ahead of the allocation is.
    ///
    /// Nothing is written here. A request stream carries no stream-type
    /// header: its first field is the leading message's own type field, which
    /// is what [`begin_request`](Self::begin_request) writes.
    async fn open_request_bi(
        &self,
    ) -> Result<(FramedSendStream, FramedRecvStream), ConnectionError> {
        let (send, recv) = self.transport.open_bi().await?;
        Ok((FramedSendStream::new(send, self.draft), FramedRecvStream::new(recv, self.draft)))
    }

    /// Reset a request stream that was opened but whose request could not be
    /// built, and pass the endpoint's error through.
    ///
    /// Without this, an endpoint refusal — the session is draining, the
    /// request-id range is exhausted — would leave a bidirectional stream
    /// open that never carries a first message, and dropping it would FIN it,
    /// telling the peer an empty stream ended cleanly.
    fn or_abandon<T>(
        halves: &mut (FramedSendStream, FramedRecvStream),
        built: Result<T, EndpointError>,
    ) -> Result<T, ConnectionError> {
        match built {
            Ok(value) => Ok(value),
            Err(e) => {
                let _ = halves.0.reset(REQUEST_CANCELLED);
                let _ = halves.1.stop(REQUEST_CANCELLED);
                Err(ConnectionError::Endpoint(e))
            }
        }
    }

    /// Write `msg` as the first message on an opened bidirectional stream and
    /// hand back the [`RequestStream`] that owns both halves.
    ///
    /// This is the one place a request reaches the wire. Every request helper
    /// funnels through it, so the ordering — open, allocate, write, emit — is
    /// stated once.
    ///
    /// A failed write resets both halves rather than leaving a half-written
    /// request stream behind. What it cannot undo is the endpoint's
    /// allocation: the request id and its state machine already exist, and
    /// there is no way to retract them, so a write that fails here leaves one
    /// pending request the endpoint will never see answered.
    async fn begin_request(
        &mut self,
        halves: (FramedSendStream, FramedRecvStream),
        kind: RequestKind,
        request_id: VarInt,
        msg: &ControlMessage,
    ) -> Result<RequestStream, ConnectionError> {
        debug_assert_eq!(
            msg.message_type(),
            kind.message_type(),
            "a request stream's first message must be the one its kind names"
        );
        let (mut send, mut recv) = halves;
        let stream_id = send.stream_id();
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Send,
            stream_kind: StreamKind::Request,
            stream_id,
        });
        let any = AnyControlMessage::Draft23(msg.clone());
        let raw = match send.write_control_with_limit(msg, self.send_control_limit).await {
            Ok(raw) => raw,
            Err(e) => {
                let _ = send.reset(REQUEST_CANCELLED);
                let _ = recv.stop(REQUEST_CANCELLED);
                return Err(e);
            }
        };
        self.emit(ClientEvent::ControlMessage {
            direction: Direction::Send,
            message: any,
            stream_id: Some(stream_id),
            raw: Some(raw),
        });
        Ok(RequestStream {
            send,
            recv,
            request_id,
            kind,
            draft: self.draft,
            stream_id,
            cancelled: false,
            finished: false,
            origin: RequestOrigin::Local,
            responded: false,
            fetch_data: None,
            // The sender of a PUBLISH is the one requester draft-23 Section
            // 6.4.2.2 excludes from "MAY FIN immediately after sending a
            // message": it is the publisher of the subscription the PUBLISH
            // establishes, so it owes a PUBLISH_DONE from the moment the
            // request goes out.
            owes_publish_done: matches!(kind, RequestKind::Publish),
        })
    }

    /// Read the next message off a request stream and dispatch it through the
    /// endpoint with that stream's own request id.
    ///
    /// On draft-23 a response carries no request id; the stream is the
    /// correlation, so the id comes from the handle and not from the wire.
    ///
    /// This blocks until a whole message has arrived. Backpressure is per
    /// request: a stream nobody reads stays unread, and the peer stays flow
    /// controlled on it alone. A peer that reset the stream surfaces as
    /// [`ConnectionError::Transport`] carrying
    /// [`TransportError::StreamReset`] with the peer's code; a caller that is
    /// deliberately not reading should watch
    /// [`RequestStream::peer_cancelled`] instead.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::Endpoint`] if the message is not one of this
    /// draft's response types, or if it does not fit the request's state.
    /// The message has already been emitted to the observer by then — what
    /// arrived is reported whether or not the endpoint accepts it.
    ///
    /// [`ConnectionError::Codec`] if the decoder refused the frame. The session is
    /// closed with the code [`Self::codec_session_error_code`] names, where it
    /// names one, as on the control stream.
    pub async fn recv_on_request_stream(
        &mut self,
        stream: &mut RequestStream,
    ) -> Result<ControlMessage, ConnectionError> {
        let capture_raw = self.observer.is_some();
        let (any, raw) =
            match stream.recv.read_control_with_limit(capture_raw, self.recv_control_limit).await {
                Ok(read) => read,
                Err(e) => {
                    // A peer that reset this stream cancelled the request on it,
                    // and this is where a caller reading normally learns of it. The
                    // record is made and its verdict dropped: the read's own error
                    // is what the caller has to act on, and returning a state error
                    // in its place would hide a reset behind it.
                    if matches!(e, ConnectionError::Transport(TransportError::StreamReset(_))) {
                        let _ = self.endpoint.cancel_request(stream.request_id);
                    }
                    let raw = stream.recv.take_refused();
                    let err = self.close_for_codec(e);
                    self.emit_refused(&err, Some(stream.stream_id()), raw);
                    return Err(err);
                }
            };
        if capture_raw {
            self.emit(ClientEvent::ControlMessage {
                direction: Direction::Receive,
                message: any.clone(),
                stream_id: Some(stream.stream_id()),
                raw: raw.clone(),
            });
        }
        let msg = match any {
            AnyControlMessage::Draft23(msg) => Ok::<_, ConnectionError>(msg),
            // `AnyControlMessage` carries one variant per enabled draft feature. With draft 23 the
            // only one enabled the arm above is exhaustive and this rejection arm unreachable.
            // Compiled in every configuration with the lint allowed, rather than gated on a `cfg`
            // naming the other drafts: such a list has to be edited in every draft module
            // whenever a draft is added, and a copy that omits one leaves this match
            // non-exhaustive.
            #[allow(unreachable_patterns)]
            _ => Err(ConnectionError::ControlMessageNarrowing),
        }?;
        // Which dispatcher this belongs to is decided by who opened the
        // stream, not by the message. On a stream this endpoint opened the
        // next message is the answer to our request; on one the peer opened it
        // cannot be, because we are the one who owes an answer. Feeding a
        // peer's REQUEST_UPDATE to the response dispatcher would look up a
        // request we never made.
        // Section 9.3: "The parameters that can appear depend on the request
        // being answered", and only the stream says which request that is.
        if let (RequestOrigin::Local, ControlMessage::RequestOk(ok)) = (stream.origin, &msg) {
            if let Some(kind) = self.endpoint.request_ok_kind(stream.request_id) {
                if let Err(e) = ok.check_answers(kind) {
                    let err = self.close_for_codec(ConnectionError::Codec(e));
                    self.emit_refused(&err, Some(stream.stream_id()), raw);
                    return Err(err);
                }
            }
        }
        let dispatched = match stream.origin {
            RequestOrigin::Local => {
                self.endpoint.receive_response_on_stream(stream.request_id, msg.clone())
            }
            RequestOrigin::Peer => {
                self.endpoint.receive_on_peer_request_stream(stream.request_id, msg.clone())
            }
        };
        if let Err(e) = dispatched {
            let err = self.close_if_session_fatal(e);
            self.emit_refused(&err, Some(stream.stream_id()), raw);
            return Err(err);
        }
        // Emitted after the dispatch rather than before it, so an observer sees
        // only a notify the endpoint accepted: one from the subscriber, or for
        // a request that is not a subscription, ends the session instead
        // (Section 9.10) and reporting it as a state change would be a lie.
        // The generic `ControlMessage` event above carries it either way.
        if let ControlMessage::PublishStateNotify(ref notify) = msg {
            self.emit(ClientEvent::PublishStateNotify {
                request_id: stream.request_id.into_inner(),
                stream_id: stream.stream_id(),
                parameters: notify.parameters.clone(),
            });
        }
        // A rejected request establishes nothing, so whatever the request type
        // would have owed on this direction is no longer owed. Without this a
        // PUBLISH the peer refused could never be finished cleanly — the
        // publisher's PUBLISH_DONE obligation would outlive the subscription
        // that was never created, and the only way out would be a reset.
        if matches!(msg, ControlMessage::RequestError(_)) {
            stream.owes_publish_done = false;
        }
        Ok(msg)
    }

    /// Write a follow-up message on an already-open request stream.
    ///
    /// The request itself was written when the stream was opened; this is for
    /// what comes after it on the same stream, PUBLISH_DONE among them — see
    /// [`publish_done`](Self::publish_done), which uses this.
    ///
    /// It does not refuse any message type. Which messages may follow a
    /// request on its own stream is not something this implementation can
    /// settle, so the choice is left to the caller rather than guessed at.
    pub async fn send_on_request_stream(
        &mut self,
        stream: &mut RequestStream,
        msg: &ControlMessage,
    ) -> Result<(), ConnectionError> {
        let any = AnyControlMessage::Draft23(msg.clone());
        let raw = stream.send.write_control_with_limit(msg, self.send_control_limit).await?;
        self.emit(ClientEvent::ControlMessage {
            direction: Direction::Send,
            message: any,
            stream_id: Some(stream.stream_id()),
            raw: Some(raw),
        });
        Ok(())
    }

    /// Cancel a request: record it at the endpoint, then terminate its stream.
    /// Section 6.4.2.3 puts the cancel at the stream — "Implementations cancel a
    /// request by abruptly terminating any directions of the stream that are
    /// still open" — while the request's own state lives in the endpoint, so
    /// the two have to move together. This is the only place that moves both.
    ///
    /// The endpoint goes first and the stream is terminated only if it agrees,
    /// which is the order every request path here uses: a caller acts on a
    /// stream after the endpoint has accepted the step, never before. A refused
    /// cancel therefore leaves the stream exactly as it was, and
    /// [`RequestStream::cancel`] is still there for a caller that wants the
    /// stream reset regardless.
    ///
    /// Idempotent from both ends: a request that has already ended accepts the
    /// cancel and stays where it is, and a handle that has already been
    /// cancelled resets nothing a second time.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::Endpoint`] if no request carries this stream's id or
    /// the request has not been written, and [`ConnectionError::Transport`] if
    /// `code` is outside the QUIC varint range — see
    /// [`RequestStream::cancel`], which is what sends it.
    pub fn cancel_request_stream(
        &mut self,
        stream: &mut RequestStream,
        code: u64,
    ) -> Result<(), ConnectionError> {
        let recorded = self.endpoint.cancel_request(stream.request_id);
        recorded.map_err(|e| self.close_if_session_fatal(e))?;
        stream.cancel(code)
    }

    /// Wait for the peer to cancel this request, and record it if it does.
    ///
    /// [`RequestStream::peer_cancelled`] with the endpoint's record attached. A
    /// caller applying backpressure is deliberately not calling
    /// [`recv_on_request_stream`](Self::recv_on_request_stream), which is the
    /// other place a peer reset surfaces, so without this the request would end
    /// on the wire and stay open in the endpoint's record for as long as the
    /// backpressure lasts.
    ///
    /// Returns what the handle's own method returns; see it for the `Ok(None)`
    /// case and for what WebTransport can and cannot observe. Cancel-safe, and
    /// it grants no flow-control credit.
    pub async fn peer_cancelled_on_request_stream(
        &mut self,
        stream: &mut RequestStream,
    ) -> Result<Option<u64>, ConnectionError> {
        let code = stream.peer_cancelled().await?;
        if code.is_some() {
            // Discarded for the reason the read path discards it: the peer has
            // ended the request whatever the record said, and a state error
            // here would replace the answer the caller asked for.
            let _ = self.endpoint.cancel_request(stream.request_id);
        }
        Ok(code)
    }

    // -- Accepting the peer's request streams -----------------------

    /// Accept the next bidirectional stream the peer opened, read the request
    /// it begins with, and hand back that request and a handle to answer it
    /// on.
    ///
    /// This is the mirror of the request helpers. Where
    /// [`subscribe`](Self::subscribe) and its siblings open a stream and write
    /// a request, this takes one the peer opened and reads one. Draft-23
    /// Section 6.3 puts requests in both directions on bidirectional streams,
    /// so a client that only ever calls the helpers can never be published to
    /// or subscribed from.
    ///
    /// The returned [`RequestStream`] carries [`RequestOrigin::Peer`]. Answer
    /// it with [`respond_subscribe_ok`](Self::respond_subscribe_ok),
    /// [`respond_ok`](Self::respond_ok) or
    /// [`respond_error`](Self::respond_error), and **hold it for as long as
    /// the request lasts** — a subscription's PUBLISH_DONE is written on it,
    /// and dropping it resets the stream.
    ///
    /// # Two refusals, two codes
    ///
    /// Draft-23 Section 6.3, on a stream that begins with the wrong type:
    /// "Bidirectional streams MUST NOT begin with any other message type
    /// unless negotiated. If they do, the peer MUST close the Session with UNEXPECTED_MESSAGE."
    /// Section 6.4.2.1, on the Request ID: "If an endpoint
    /// receives a Request ID where the least significant bit is incorrect for the sender, it MUST
    /// close the session with INVALID_REQUEST_ID. If an endpoint detects a duplicate Request ID, it
    /// MUST close the session with INVALID_REQUEST_ID." Both are closes of the session on the wire,
    /// with
    /// different codes, and both happen before this returns — the error handed
    /// back reports a session that is already gone, not one the caller must
    /// remember to close.
    ///
    /// # Cancelling this future loses nothing
    ///
    /// A stream taken off the transport but not yet read is put back on an
    /// internal queue, and the next call takes it before accepting anything
    /// new — including whatever bytes of the request had already arrived,
    /// which live in the stream's own reader. So this is safe to `select!`
    /// against a shutdown signal or a timer. See
    /// [`pending_inbound_count`](Self::pending_inbound_count).
    ///
    /// What it is **not** safe to do is run concurrently with another method
    /// on the same connection: this takes `&mut self` because registering the
    /// peer's request moves endpoint state, and no signature avoids that while
    /// the connection owns the endpoint. A caller blocked in
    /// [`recv_on_request_stream`](Self::recv_on_request_stream) waiting for
    /// its own response is not accepting, and the peer's request streams queue
    /// up in the transport behind it. One loop that never blocks indefinitely
    /// on a single read is the shape this supports.
    ///
    /// # Ordering
    ///
    /// The endpoint is told about the request last, after every step that can
    /// fail or be cancelled, and building the handle afterwards cannot fail.
    /// This is the inverse of the outbound path's reasoning — it opens the
    /// stream before allocating a Request ID for the same reason — and rests
    /// on the same fact: the endpoint has no way to abandon a request it has
    /// already registered. Registering earlier would let a cancelled accept
    /// leave a state machine keyed to a stream nobody holds, and the peer's
    /// next use of that Request ID would then be reported as a duplicate — a
    /// session close, over an id the peer used exactly once.
    ///
    /// # Errors
    ///
    /// - [`ConnectionError::NonRequestOnRequestStream`] — the session has been
    ///   closed with UNEXPECTED_MESSAGE and the stream reset.
    /// - [`ConnectionError::Endpoint`] carrying `RequestId` or
    ///   `DuplicateRequestId` — the session has been closed with
    ///   INVALID_REQUEST_ID and the stream reset.
    /// - [`ConnectionError::Endpoint`] carrying `NotActive` or `Draining` —
    ///   the stream is reset, the session is left alone.
    /// - [`ConnectionError::Transport`] — the stream is reset, the session is
    ///   left alone.
    /// - [`ConnectionError::Codec`] — the stream is reset, and the session is
    ///   closed with the code [`Self::codec_session_error_code`] names, where it
    ///   names one. Section 9: "Every message on a control or request stream is
    ///   formatted as follows", so the control message rules hold here too.
    pub async fn accept_request_stream(
        &mut self,
    ) -> Result<(ControlMessage, RequestStream), ConnectionError> {
        let pair = match self.take_pending_inbound() {
            Some(pair) => pair,
            None => {
                let (send, recv) = self.transport.accept_bi().await?;
                (FramedSendStream::new(send, self.draft), FramedRecvStream::new(recv, self.draft))
            }
        };
        let capture_raw = self.observer.is_some();

        let (any, raw, mut send, mut recv) = {
            let mut pending = PendingInbound { pair: Some(pair), queue: &self.pending_inbound };
            let read = {
                let (_, recv) = pending.pair.as_mut().expect("set on construction");
                recv.read_control_with_limit(capture_raw, self.recv_control_limit).await
            };
            // Taken out before anything can return, so the guard's Drop puts
            // the pair back for exactly one reason: this future was cancelled.
            let (mut send, mut recv) = pending.pair.take().expect("set on construction");
            match read {
                Ok((any, raw)) => (any, raw, send, recv),
                Err(e) => {
                    // A stream whose first message could not be read is not
                    // worth queueing: the next accept would fail on it the
                    // same way. Reset rather than FIN — nothing was served.
                    let _ = send.reset(REQUEST_UNANSWERED);
                    let _ = recv.stop(REQUEST_UNANSWERED);
                    let raw = recv.take_refused();
                    let err = self.close_for_codec(e);
                    self.emit_refused(&err, Some(send.stream_id()), raw);
                    return Err(err);
                }
            }
        };

        // Reported once the request has actually arrived rather than when the
        // stream came off the transport, so a cancelled accept that is retried
        // does not report the same stream twice.
        let stream_id = send.stream_id();
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Receive,
            stream_kind: StreamKind::Request,
            stream_id,
        });
        if capture_raw {
            self.emit(ClientEvent::ControlMessage {
                direction: Direction::Receive,
                message: any.clone(),
                stream_id: Some(stream_id),
                raw: raw.clone(),
            });
        }

        let msg = match any {
            AnyControlMessage::Draft23(msg) => msg,
            // `AnyControlMessage` carries one variant per enabled draft feature. With draft 23 the
            // only one enabled the arm above is exhaustive and this rejection arm unreachable.
            // Compiled in every configuration with the lint allowed, rather than gated on a `cfg`
            // naming the other drafts: such a list has to be edited in every draft module
            // whenever a draft is added, and a copy that omits one leaves this match
            // non-exhaustive.
            #[allow(unreachable_patterns)]
            _ => {
                let _ = send.reset(REQUEST_UNANSWERED);
                let _ = recv.stop(REQUEST_UNANSWERED);
                return Err(ConnectionError::ControlMessageNarrowing);
            }
        };

        let ty = msg.message_type();
        let Some(kind) = RequestKind::from_message_type(ty) else {
            let err = self.endpoint.refuse_non_request(ty);
            self.close_for(&err);
            let _ = send.reset(REQUEST_UNANSWERED);
            let _ = recv.stop(REQUEST_UNANSWERED);
            let refused = ConnectionError::NonRequestOnRequestStream(ty);
            self.emit_refused(&refused, Some(stream_id), raw);
            return Err(refused);
        };

        let request_id = match self.endpoint.receive_request_on_stream(&msg) {
            Ok(request_id) => request_id,
            Err(e) => {
                let _ = send.reset(REQUEST_UNANSWERED);
                let _ = recv.stop(REQUEST_UNANSWERED);
                let err = self.close_if_session_fatal(e);
                self.emit_refused(&err, Some(stream_id), raw);
                return Err(err);
            }
        };

        Ok((
            msg,
            RequestStream {
                send,
                recv,
                request_id,
                kind,
                draft: self.draft,
                stream_id,
                cancelled: false,
                finished: false,
                origin: RequestOrigin::Peer,
                responded: false,
                fetch_data: None,
                // A peer's SUBSCRIBE does not establish a subscription until
                // it is accepted, so nothing is owed until
                // `respond_subscribe_ok` runs.
                owes_publish_done: false,
            },
        ))
    }

    /// Take the oldest stream pair a cancelled
    /// [`accept_request_stream`](Self::accept_request_stream) put back, if any.
    ///
    /// Synchronous on purpose, like
    /// [`take_deferred_uni`](Self::take_deferred_uni): the guard is dropped
    /// before the caller awaits, so the lock is never held across a suspension
    /// point.
    fn take_pending_inbound(&self) -> Option<(FramedSendStream, FramedRecvStream)> {
        self.pending_inbound.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).pop_front()
    }

    /// How many peer-opened request streams a cancelled
    /// [`accept_request_stream`](Self::accept_request_stream) put back and a
    /// later call has not yet taken.
    ///
    /// Zero unless an accept future was dropped mid-read.
    pub fn pending_inbound_count(&self) -> usize {
        self.pending_inbound.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).len()
    }

    // -- Answering the peer's requests ------------------------------

    /// Write `msg` on the request stream `stream` carries, driving the
    /// endpoint first and the wire second.
    ///
    /// The response goes on the request's own bidirectional stream and never
    /// on the control stream: draft-23 responses carry no Request ID, so the
    /// stream is the only thing that says what is being answered. Taking the
    /// id off the handle rather than from the caller makes that correlation
    /// unforgeable.
    ///
    /// `fin` is true only for REQUEST_ERROR. See
    /// [`respond_error`](Self::respond_error).
    async fn respond(
        &mut self,
        stream: &mut RequestStream,
        msg: ControlMessage,
        fin: bool,
    ) -> Result<(), ConnectionError> {
        // A request this endpoint made is answered by the peer, with one
        // exception draft-23 Section 9.5 states outright: "A subscriber can
        // also send REQUEST_UPDATE to modify parameters of a subscription
        // established with PUBLISH", and the receiver of one "MUST respond with
        // exactly one REQUEST_UPDATE_OK or REQUEST_UPDATE_ERROR message
        // indicating if the update was successful" — Section 1.5's names for a
        // REQUEST_OK and a REQUEST_ERROR answering an update. On a PUBLISH this
        // endpoint sent, that receiver is this endpoint, so the one response it
        // may write on a stream of its own is the answer to an update waiting
        // there.
        let answers_an_update =
            matches!(msg, ControlMessage::RequestOk(_) | ControlMessage::RequestError(_))
                && self.endpoint.has_unanswered_update(stream.request_id);
        if stream.origin != RequestOrigin::Peer && !answers_an_update {
            return Err(ConnectionError::RespondedToOwnRequest(stream.request_id.into_inner()));
        }
        // The endpoint first, so a response that does not fit the request's
        // state is refused before any of it reaches the wire. What this cannot
        // undo is a write that fails afterwards, which leaves the state
        // machine one step ahead of the peer — the same asymmetry
        // `begin_request` carries on the outbound side.
        self.endpoint.send_response_on_stream(stream.request_id, &msg)?;
        self.send_on_request_stream(stream, &msg).await?;
        stream.responded = true;
        // A REQUEST_ERROR answering an update retires nothing: the
        // subscription is live and Section 9.5 now asks for a PUBLISH_DONE
        // carrying UPDATE_FAILED. So neither the flag nor the send half is
        // cleared, and the obligation this draft already models goes on
        // standing until that message is written.
        // Section 9.5.1: "When a REQUEST_UPDATE fails for a FETCH, the
        // publisher MUST reset the FETCH data stream." A REQUEST_ERROR on a
        // fetch that has already been answered can only be answering an
        // update, because the request's own answer was the FETCH_OK; one
        // before that refuses the fetch itself, and there is no data stream
        // open to reset.
        if fin && stream.kind == RequestKind::Fetch && stream.responded {
            stream.reset_fetch_data();
        }
        if fin && !self.endpoint.owes_update_failure(stream.request_id) {
            // A rejected request owes nothing further, so the guard in
            // `finish` cannot be standing: clear the flag rather than let a
            // REQUEST_ERROR trip over an obligation the error just retired.
            stream.owes_publish_done = false;
            stream.finish().await?;
        }
        Ok(())
    }

    /// Answer a peer's PUBLISH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE,
    /// SUBSCRIBE_TRACKS or TRACK_STATUS with REQUEST_OK.
    ///
    /// The one success message for every request kind but SUBSCRIBE: draft-23
    /// Section 9.3, "The REQUEST_OK message is sent in response to PUBLISH,
    /// FETCH, REQUEST_UPDATE, TRACK_STATUS, SUBSCRIBE_NAMESPACE,
    /// SUBSCRIBE_TRACKS and PUBLISH_NAMESPACE requests."
    ///
    /// The send half is left open. A SUBSCRIBE_NAMESPACE responder still owes
    /// the peer the namespaces it accepted, so finishing here would end the
    /// request before it had been served; a TRACK_STATUS or FETCH responder
    /// owes nothing further on this stream and may call
    /// [`RequestStream::finish`] straight after.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::RespondedToOwnRequest`] if `stream` is one this
    /// endpoint opened, and [`ConnectionError::Endpoint`] if no request of a
    /// kind REQUEST_OK answers is pending on it — including the Section 9.3
    /// case of Track Properties on a response other than TRACK_STATUS_OK and
    /// FETCH_OK.
    /// Nothing is written in any of them.
    pub async fn respond_ok(
        &mut self,
        stream: &mut RequestStream,
        response: RequestOk,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::RequestOk(response), false).await
    }

    /// Answer a peer's SUBSCRIBE with SUBSCRIBE_OK.
    ///
    /// The send half is left open, and it must be: this endpoint is now the
    /// publisher of an established subscription and owes it a PUBLISH_DONE,
    /// which travels on this same stream —
    /// [`publish_done_on`](Self::publish_done_on). Until that has been sent,
    /// [`RequestStream::finish`] refuses, per draft-23 Section 6.4.2.2.
    pub async fn respond_subscribe_ok(
        &mut self,
        stream: &mut RequestStream,
        response: SubscribeOk,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::SubscribeOk(response), false).await?;
        stream.owes_publish_done = true;
        Ok(())
    }

    /// Reject a peer's request with REQUEST_ERROR, and finish the send half.
    ///
    /// The FIN is part of the act, not a convenience: draft-23 Section 6.4.2.3
    /// says "When an endpoint rejects a request without performing any
    /// application processing, it SHOULD send a REQUEST_ERROR and FIN the
    /// stream." It is also the one response that can be finished immediately,
    /// because a rejected request leaves nothing further to send — every
    /// success path owes the peer something more, which is what Section 6.4.2.2
    /// makes a MUST NOT for the rest of them.
    ///
    /// A finished handle does nothing further on [`Drop`], so the rejected
    /// stream is not then reset.
    pub async fn respond_error(
        &mut self,
        stream: &mut RequestStream,
        response: RequestError,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::RequestError(response), true).await
    }

    /// End a subscription this endpoint accepted, on the stream the peer's
    /// SUBSCRIBE opened.
    ///
    /// The mirror of [`publish_done`](Self::publish_done), which ends a
    /// publication this endpoint offered with PUBLISH. Both write PUBLISH_DONE
    /// on a request stream and take the Request ID off the handle; they differ
    /// in which state machine moves, and therefore in which one refuses.
    ///
    /// The send half is finished once the message is out, for the reason
    /// [`publish_done`](Self::publish_done) gives: Section 6.4.2.2 names
    /// PUBLISH_DONE as the last required message on a publisher's direction and
    /// asks for the FIN promptly after it.
    pub async fn publish_done_on(
        &mut self,
        stream: &mut RequestStream,
        status_code: VarInt,
        stream_count: VarInt,
        reason_phrase: Vec<u8>,
    ) -> Result<(), ConnectionError> {
        let msg = ControlMessage::PublishDone(moqtap_codec::draft23::message::PublishDone {
            status_code,
            stream_count,
            reason_phrase,
        });
        self.respond(stream, msg, false).await?;
        stream.owes_publish_done = false;
        stream.finish().await
    }

    /// Announce a namespace on a peer's SUBSCRIBE_NAMESPACE stream.
    ///
    /// Draft-23 Section 9.15 puts NAMESPACE "on the response stream of a
    /// SUBSCRIBE_NAMESPACE request", and Table 5 gives it the Stream value
    /// Request rather than Control — which is why this exists at all and why
    /// [`send_control`](Self::send_control) is the wrong route for it. The
    /// announcements follow the acceptance — Section 4.2: "On success, the
    /// publisher MUST send a NAMESPACE message for each namespace it knows
    /// that matches the Track Namespace Prefix" — so this refuses until
    /// [`respond_ok`](Self::respond_ok) has run on the same stream.
    pub async fn namespace_on(
        &mut self,
        stream: &mut RequestStream,
        response: Namespace,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::Namespace(response), false).await
    }

    /// Withdraw a namespace on a peer's SUBSCRIBE_NAMESPACE stream.
    ///
    /// Section 4.2: "The publisher MUST NOT send NAMESPACE_DONE for a
    /// namespace suffix before the corresponding NAMESPACE." Which suffixes
    /// have been announced is the caller's bookkeeping, not this connection's,
    /// so what is enforced here is only that the request was accepted first.
    pub async fn namespace_done_on(
        &mut self,
        stream: &mut RequestStream,
        response: NamespaceDone,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::NamespaceDone(response), false).await
    }

    /// Tell the peer a track matching its SUBSCRIBE_TRACKS will not be
    /// published, on that request's own stream.
    ///
    /// Section 9.18: "All PUBLISH_SKIPPED messages are in response to a
    /// SUBSCRIBE_TRACKS," which is why this refuses any other kind of stream —
    /// the endpoint looks the Request ID up among the SUBSCRIBE_TRACKS
    /// requests alone.
    pub async fn publish_skipped_on(
        &mut self,
        stream: &mut RequestStream,
        response: PublishSkipped,
    ) -> Result<(), ConnectionError> {
        self.respond(stream, ControlMessage::PublishSkipped(response), false).await
    }

    // -- Subscribe flow ---------------------------------------------

    /// Send a SUBSCRIBE on a bidirectional stream of its own.
    ///
    /// The returned [`RequestStream`] is where SUBSCRIBE_OK, REQUEST_ERROR
    /// and later PUBLISH_DONE arrive — read them with
    /// [`recv_on_request_stream`](Self::recv_on_request_stream). **Hold it for
    /// the subscription's life**: dropping it resets the stream, which
    /// cancels the subscription.
    pub async fn subscribe(
        &mut self,
        track_namespace: TrackNamespace,
        track_name: Vec<u8>,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.subscribe(track_namespace, track_name, parameters),
        )?;
        self.begin_request(halves, RequestKind::Subscribe, req_id, &msg).await
    }

    // // Draft-23 has no UNSUBSCRIBE. Subscribers end a
    // subscription by resetting its request stream — `RequestStream::cancel`
    // — or wait for PublishDone.

    // -- Fetch flow -------------------------------------------------

    /// Send a FETCH on a bidirectional stream of its own.
    ///
    /// FETCH_OK or REQUEST_ERROR comes back on the returned
    /// [`RequestStream`]; the fetched objects arrive on separate
    /// unidirectional data streams. Dropping the handle cancels the fetch.
    ///
    /// # The range is a parameter now
    ///
    /// Draft-23 Section 9.11 has no `Fetch Type`, no Standalone or Joining structure, and no inline
    /// `Start Location` / `End Location` pair. The range travels in the `LOCATION_FILTER`
    /// parameter, so this takes the same three arguments
    /// [`subscribe`](Self::subscribe) does. [`fetch_range`](Self::fetch_range)
    /// is the same call with the filter built for you; a FETCH with no filter
    /// covers `{0,0}` through Largest Object, inclusive.
    pub async fn fetch(
        &mut self,
        track_namespace: TrackNamespace,
        track_name: Vec<u8>,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.fetch(track_namespace, track_name, parameters),
        )?;
        self.begin_request(halves, RequestKind::Fetch, req_id, &msg).await
    }

    /// Send a FETCH for one range, carried as a `LOCATION_FILTER` parameter.
    ///
    /// **The range is inclusive at both ends** (draft-23 Section 3.2 and
    /// draft-23 Section 3.3.1), and nothing here adds or subtracts one.
    pub async fn fetch_range(
        &mut self,
        track_namespace: TrackNamespace,
        track_name: Vec<u8>,
        range: &LocationFilter,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.fetch_range(track_namespace, track_name, range, parameters),
        )?;
        self.begin_request(halves, RequestKind::Fetch, req_id, &msg).await
    }

    // Draft-23 has no joining fetch. Section 9.11 has no Fetch Type field, no variant structures
    // and no joining mechanism, and
    // Section 3.4's fill fetch stream replaces it: a subscriber asks for the
    // Objects behind the live edge by putting a `FILL_PARAMETERS` parameter on
    // its SUBSCRIBE (or on a later REQUEST_UPDATE) rather than by opening a
    // second request. See [`accept_fill_stream`](Self::accept_fill_stream) for
    // the stream that answers it, and `crate::draft23::fill` for the parameter.

    // // Draft-23 has no FETCH_CANCEL. Fetchers abort
    // with `RequestStream::cancel`, which resets the request stream.

    // -- Namespace flows --------------------------------------------

    /// Send a SUBSCRIBE_NAMESPACE on a bidirectional stream of its own.
    ///
    /// This call sends SUBSCRIBE_NAMESPACE (type 0x50), which subscribes to NAMESPACE /
    /// NAMESPACE_DONE announcements only. To receive PUBLISH messages for
    /// matching tracks, use [`Self::subscribe_tracks`] instead — a separate
    /// request on a request stream of its own.
    pub async fn subscribe_namespace(
        &mut self,
        namespace_prefix: TrackNamespace,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.subscribe_namespace(namespace_prefix, parameters),
        )?;
        self.begin_request(halves, RequestKind::SubscribeNamespace, req_id, &msg).await
    }

    /// Send a SUBSCRIBE_TRACKS (type 0x51) on a bidirectional
    /// stream of its own. Causes the relay to PUBLISH matching tracks back to
    /// us.
    ///
    /// One of the seven request types draft-23 Section 6.3 lists.
    pub async fn subscribe_tracks(
        &mut self,
        namespace_prefix: TrackNamespace,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.subscribe_tracks(namespace_prefix, parameters),
        )?;
        self.begin_request(halves, RequestKind::SubscribeTracks, req_id, &msg).await
    }

    /// Send a PUBLISH_NAMESPACE on a bidirectional stream of its own.
    pub async fn publish_namespace(
        &mut self,
        track_namespace: TrackNamespace,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.publish_namespace(track_namespace, parameters),
        )?;
        self.begin_request(halves, RequestKind::PublishNamespace, req_id, &msg).await
    }

    // -- Track Status flow ------------------------------------------

    /// Send a TRACK_STATUS on a bidirectional stream of its own.
    pub async fn track_status(
        &mut self,
        track_namespace: TrackNamespace,
        track_name: Vec<u8>,
        parameters: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.track_status(track_namespace, track_name, parameters),
        )?;
        self.begin_request(halves, RequestKind::TrackStatus, req_id, &msg).await
    }

    // -- Publish flow (publisher side) ------------------------------

    /// Send a PUBLISH on a bidirectional stream of its own.
    ///
    /// REQUEST_OK — which is what draft-23 answers a PUBLISH with, PUBLISH_OK
    /// having been folded into it — or REQUEST_ERROR comes back on the
    /// returned [`RequestStream`], and [`publish_done`](Self::publish_done) is
    /// written back on it when the publication ends, so the handle must be
    /// held for as long as the publication lasts.
    pub async fn publish(
        &mut self,
        track_namespace: TrackNamespace,
        track_name: Vec<u8>,
        track_alias: VarInt,
        parameters: Vec<KeyValuePair>,
        track_properties: Vec<KeyValuePair>,
    ) -> Result<RequestStream, ConnectionError> {
        let mut halves = self.open_request_bi().await?;
        let (req_id, msg) = Self::or_abandon(
            &mut halves,
            self.endpoint.publish(
                track_namespace,
                track_name,
                track_alias,
                parameters,
                track_properties,
            ),
        )?;
        self.begin_request(halves, RequestKind::Publish, req_id, &msg).await
    }

    /// Send a PUBLISH_DONE on the request stream the PUBLISH opened.
    ///
    /// PUBLISH_DONE is a response and carries no request id on the wire, so
    /// the stream is the only thing that says which publication ended. The id
    /// the endpoint needs is taken off `stream`, which makes the correlation
    /// unforgeable — there is no way to name one request and write on
    /// another's stream.
    ///
    /// The send half is finished once the message is out. Draft-23 Section
    /// 6.4.2.2 names PUBLISH_DONE as the last required message on a publisher's
    /// direction — "the publisher of an Established subscription MUST send
    /// PUBLISH_DONE, before sending a FIN" — and asks for the FIN promptly
    /// afterwards, since nothing further is owed on that direction. Leaving it
    /// open instead would end the publication at [`Drop`], which resets the
    /// stream with REQUEST_CANCELLED and tells the peer the request was
    /// abandoned rather than completed.
    ///
    /// The receive half stays open. A FIN closes one direction, so the
    /// subscriber can still be heard from on its own.
    pub async fn publish_done(
        &mut self,
        stream: &mut RequestStream,
        status_code: VarInt,
        stream_count: VarInt,
        reason_phrase: Vec<u8>,
    ) -> Result<(), ConnectionError> {
        let request_id = stream.request_id();
        let msg = self.endpoint.send_publish_done(
            request_id,
            status_code,
            stream_count,
            reason_phrase,
        )?;
        self.send_on_request_stream(stream, &msg).await?;
        stream.owes_publish_done = false;
        stream.finish().await
    }

    // -- Data streams -----------------------------------------------

    /// Open a new unidirectional stream for sending subgroup data.
    pub async fn open_subgroup_stream(
        &self,
        header: &AnySubgroupHeader,
    ) -> Result<FramedSendStream, ConnectionError> {
        let send = self.transport.open_uni().await?;
        let mut framed = FramedSendStream::new(send, self.draft);
        let sid = framed.stream_id();
        framed.write_subgroup_header(header).await?;
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Send,
            stream_kind: StreamKind::Subgroup,
            stream_id: sid,
        });
        self.emit(ClientEvent::DataStreamHeader {
            stream_id: sid,
            direction: Direction::Send,
            header: header.clone(),
        });
        Ok(framed)
    }

    /// Open a new unidirectional stream for sending a FETCH's objects.
    ///
    /// The objects answering a FETCH do not go on the request's own stream:
    /// they go on a unidirectional stream of their own, which opens with a
    /// FETCH_HEADER naming the request they belong to. This writes that header
    /// and hands back the stream, the same way
    /// [`open_subgroup_stream`](Self::open_subgroup_stream) does for a
    /// subgroup.
    ///
    /// The caller owns the stream that comes back. Nothing here remembers
    /// which request it belongs to, so an endpoint serving several fetches at
    /// once keeps its own map from Request ID to stream.
    pub async fn open_fetch_stream(
        &self,
        header: &AnyFetchHeader,
    ) -> Result<FramedSendStream, ConnectionError> {
        let send = self.transport.open_uni().await?;
        let mut framed = FramedSendStream::new(send, self.draft);
        let sid = framed.stream_id();
        framed.write_fetch_header(header).await?;
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Send,
            stream_kind: StreamKind::Fetch,
            stream_id: sid,
        });
        Ok(framed)
    }

    /// Open the data stream for a FETCH this endpoint is answering, and keep
    /// the handle on the request.
    ///
    /// The same stream [`open_fetch_stream`](Self::open_fetch_stream) returns,
    /// parked on the request stream it belongs to. That is what lets a rule
    /// about the fetch reach the objects it is serving: a refused
    /// REQUEST_UPDATE has to reset this stream, and the connection cannot
    /// reset a handle the caller walked away with.
    ///
    /// Write objects through
    /// [`RequestStream::fetch_data`](RequestStream::fetch_data), or through
    /// the borrow this returns.
    pub async fn open_fetch_stream_on<'s>(
        &self,
        stream: &'s mut RequestStream,
        header: &AnyFetchHeader,
    ) -> Result<&'s mut FramedSendStream, ConnectionError> {
        let framed = self.open_fetch_stream(header).await?;
        stream.fetch_data = Some(framed);
        Ok(stream.fetch_data.as_mut().expect("just stored"))
    }

    /// Accept an incoming unidirectional data stream and read its subgroup
    /// header.
    ///
    /// Streams the peer opened before its control stream are returned first,
    /// in arrival order, before any new one is accepted from the transport:
    /// [`connect`](Self::connect) had to look at them to find the control
    /// stream and set the rest aside rather than drop them. They are
    /// otherwise ordinary — the type varint `connect` read is still on the
    /// front of each one.
    ///
    /// A FETCH_HEADER stream, fetch or fill, is never read as a subgroup: it
    /// is set aside for [`accept_fetch_stream`](Self::accept_fetch_stream)
    /// and [`accept_fill_stream`](Self::accept_fill_stream), and this waits
    /// for the next stream.
    pub async fn accept_subgroup_stream(
        &self,
    ) -> Result<(AnySubgroupHeader, FramedRecvStream), ConnectionError> {
        let mut framed = self.next_non_fetch_uni().await?;
        let sid = framed.stream_id();
        let header = framed.read_subgroup_header().await?;
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Receive,
            stream_kind: StreamKind::Subgroup,
            stream_id: sid,
        });
        self.emit(ClientEvent::DataStreamHeader {
            stream_id: sid,
            direction: Direction::Receive,
            header: header.clone(),
        });
        // The track is resolved here and not inside the stream: it takes the
        // endpoint's alias table, which a stream handle has no way back to.
        // Handed over rather than offered, so measuring is not something a
        // caller has to remember to ask for.
        if let (Some(objects), Some(existence)) = (
            self.endpoint.track_objects(header.track_alias()),
            self.endpoint.track_existence(header.track_alias()),
        ) {
            framed.measure_objects_against(objects, existence, header.group_id());
        }
        Ok((header, framed))
    }

    /// Accept the next unidirectional stream and read its fetch header.
    ///
    /// [`accept_subgroup_stream`](Self::accept_subgroup_stream)'s twin. The two
    /// are separate because the header decides how every object after it is
    /// framed, so a caller has to know which it is expecting before the first
    /// byte is read.
    ///
    /// Objects come off the returned stream with
    /// [`FramedRecvStream::read_fetch_object`].
    ///
    /// FETCH_HEADER streams another accept set aside come first, in arrival
    /// order. A stream of any other type that arrives while this waits is
    /// set aside for [`accept_subgroup_stream`](Self::accept_subgroup_stream).
    pub async fn accept_fetch_stream(
        &self,
    ) -> Result<(AnyFetchHeader, FramedRecvStream), ConnectionError> {
        let mut framed = self.next_fetch_uni().await?;
        let sid = framed.stream_id();
        let header = framed.read_fetch_header().await?;
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Receive,
            stream_kind: StreamKind::Fetch,
            stream_id: sid,
        });
        self.emit(ClientEvent::FetchStreamHeader {
            stream_id: sid,
            direction: Direction::Receive,
            header: header.clone(),
        });
        // A fetch header goes out as `FetchStreamHeader`; `DataStreamHeader`
        // carries an `AnySubgroupHeader` and cannot express one.
        framed.fetch_existence = self.endpoint.fetch_existence(header.request_id());
        Ok((header, framed))
    }

    /// Accept the next unidirectional stream as a **fill fetch stream** (Section 3.4).
    ///
    /// A fill fetch stream is framed exactly as a fetch response — a
    /// FETCH_HEADER and the fetch object records after it, read with
    /// [`FramedRecvStream::read_fetch_object`] — and differs only in what its
    /// Request ID names: the SUBSCRIBE (or the REQUEST_UPDATE, which travels
    /// under the same Request ID) that asked for the fill, rather than a FETCH.
    /// So this is [`accept_fetch_stream`](Self::accept_fetch_stream) with the
    /// attribution done: the endpoint is asked whether that request asked for a
    /// fill, the stream is counted against the subscription, and the event goes
    /// out under [`StreamKind::Fill`].
    ///
    /// # The object reader is already started
    ///
    /// Unlike [`accept_fetch_stream`](Self::accept_fetch_stream), the returned
    /// stream needs no
    /// [`begin_fetch_objects`](FramedRecvStream::begin_fetch_objects) call:
    /// objects can be read straight off it. A fetch stream cannot be started
    /// here because the Group Order lives on the FETCH the caller sent and
    /// nothing on this side holds it; a **fill** stream can, because the
    /// endpoint kept the order when the subscription asked for the fill — see
    /// [`Endpoint::fill_group_order`](crate::draft23::endpoint::Endpoint::fill_group_order)
    /// and [`fill::group_order`](crate::draft23::fill::group_order) for the
    /// three-step resolution Section 3.4 and Section 9.19.8 define.
    ///
    /// Doing it here rather than leaving it to the caller is the difference
    /// between an error and a wrong answer: a reader started in the wrong
    /// direction still parses every frame, and reports Group IDs that walk the
    /// wrong way. A caller that has since learned better restarts it by calling
    /// `begin_fetch_objects` itself before reading the first object.
    ///
    /// # What ends it, and what that does not do
    ///
    /// Section 3.4.1: a FIN means the whole fill range was delivered, and a
    /// reset means the fill failed — "Because there is no REQUEST_ERROR
    /// associated with a fill fetch stream, the publisher signals a fill
    /// failure by resetting the stream". Neither touches the subscription:
    /// "Resetting or cancelling a fill fetch stream, by either endpoint, does
    /// not affect the subscription, which continues to deliver objects using
    /// subscribe subgroups and datagrams." Tell the endpoint which happened
    /// with [`Endpoint::on_fill_stream_ended`](crate::draft23::endpoint::Endpoint::on_fill_stream_ended);
    /// a subscriber that wants to stop one early sends `STOP_SENDING` with
    /// [`FramedRecvStream::stop`], which Section 3.4.1 gives it outright.
    ///
    /// # Errors
    ///
    /// [`ConnectionError::Endpoint`] carrying
    /// [`EndpointError::UnrequestedFillStream`] when the header names a request
    /// that asked for no fill. Not fatal to the session — draft-23 states no
    /// rule for the case, and the stream is the only thing wrong.
    ///
    /// # Which stream it takes
    ///
    /// The same one [`accept_fetch_stream`](Self::accept_fetch_stream)
    /// would: a FETCH_HEADER stream another accept set aside, else the next
    /// one to arrive. Streams of any other type are set aside for
    /// [`accept_subgroup_stream`](Self::accept_subgroup_stream).
    pub async fn accept_fill_stream(
        &mut self,
    ) -> Result<(AnyFetchHeader, FramedRecvStream), ConnectionError> {
        let mut framed = self.next_fetch_uni().await?;
        let sid = framed.stream_id();
        let header = framed.read_fetch_header().await?;
        let request_id = VarInt::from_u64(header.request_id())
            .map_err(|e| ConnectionError::Codec(CodecError::VarInt(e)))?;
        self.endpoint.on_fill_stream_opened(request_id)?;
        // `on_fill_stream_opened` returned Ok, so the request asked for a fill
        // and the entry that answers this exists; the fallback is what a future
        // caller that split the two lookups would get, and it is the same
        // Section 9.19.8 default `fill::group_order` applies.
        let group_order =
            self.endpoint.fill_group_order(request_id).unwrap_or(GroupOrder::Ascending);
        framed.begin_fetch_objects(group_order);
        self.emit(ClientEvent::StreamOpened {
            direction: Direction::Receive,
            stream_kind: StreamKind::Fill,
            stream_id: sid,
        });
        self.emit(ClientEvent::FetchStreamHeader {
            stream_id: sid,
            direction: Direction::Receive,
            header: header.clone(),
        });
        Ok((header, framed))
    }

    /// Take the oldest stream [`connect`](Self::connect) set aside, if any.
    ///
    /// Synchronous on purpose: the guard is dropped before the caller awaits,
    /// so the lock is never held across a suspension point. A poisoned lock
    /// is recovered rather than propagated — nothing here can leave the queue
    /// in a state a later reader could be misled by, since the only mutations
    /// are a `push_back` and a `pop_front`.
    fn take_deferred_uni(&self) -> Option<FramedRecvStream> {
        self.deferred_uni.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).pop_front()
    }

    /// Take the oldest FETCH_HEADER stream set aside, if any. Synchronous for
    /// the reason [`take_deferred_uni`](Self::take_deferred_uni) is.
    fn take_deferred_fetch(&self) -> Option<FramedRecvStream> {
        self.deferred_fetch.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).pop_front()
    }

    fn set_aside(queue: &Mutex<VecDeque<FramedRecvStream>>, framed: FramedRecvStream) {
        queue.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).push_back(framed);
    }

    /// The next stream that is not a FETCH_HEADER stream: a set-aside one
    /// first, then the transport's. FETCH_HEADER streams met on the way are
    /// set aside. A stream whose type cannot be read is returned, so that the
    /// header read reports why.
    async fn next_non_fetch_uni(&self) -> Result<FramedRecvStream, ConnectionError> {
        loop {
            let mut framed = match self.take_deferred_uni() {
                Some(framed) => framed,
                None => FramedRecvStream::new(self.transport.accept_uni().await?, self.draft),
            };
            match framed.peek_stream_type().await {
                Ok(FETCH_STREAM_TYPE) => Self::set_aside(&self.deferred_fetch, framed),
                _ => return Ok(framed),
            }
        }
    }

    /// The next FETCH_HEADER stream: a set-aside one first, then the
    /// transport's. Streams of other types met on the way are set aside. A
    /// stream whose type cannot be read is returned, so that the header read
    /// reports why.
    async fn next_fetch_uni(&self) -> Result<FramedRecvStream, ConnectionError> {
        if let Some(framed) = self.take_deferred_fetch() {
            return Ok(framed);
        }
        loop {
            let mut framed = FramedRecvStream::new(self.transport.accept_uni().await?, self.draft);
            match framed.peek_stream_type().await {
                Ok(stream_type) if stream_type != FETCH_STREAM_TYPE => {
                    Self::set_aside(&self.deferred_uni, framed)
                }
                _ => return Ok(framed),
            }
        }
    }

    /// How many unidirectional streams have been set aside and not yet handed
    /// back: those [`connect`](Self::connect) met before the control stream,
    /// and those an accept met that were for another accept.
    ///
    /// Zero for a peer that opened its control stream first and a caller
    /// that accepted each stream as the kind it is, which is the ordinary
    /// case.
    pub fn deferred_stream_count(&self) -> usize {
        self.deferred_uni.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).len()
            + self.deferred_fetch.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).len()
    }

    /// Send an object via datagram.
    ///
    /// A header whose object status the framing cannot carry is refused rather
    /// than sent. A datagram writes a status field only when its type byte sets
    /// the STATUS bit, so a header carrying End of Group under a type byte
    /// without that bit would go out as an ordinary payload object with the
    /// marker missing — a datagram the peer cannot tell from one that never had
    /// a status. `AnyDatagramHeader::encode` answers that with
    /// `CodecError::InvalidField` and writes nothing, on every draft.
    pub fn send_datagram(
        &self,
        header: &AnyDatagramHeader,
        payload: &[u8],
    ) -> Result<(), ConnectionError> {
        let mut buf = Vec::new();
        header.encode(&mut buf)?;
        buf.extend_from_slice(payload);
        self.emit(ClientEvent::DatagramReceived {
            direction: Direction::Send,
            header: header.clone(),
            payload_len: payload.len(),
        });
        self.transport.send_datagram(bytes::Bytes::from(buf))?;
        Ok(())
    }

    /// Receive a datagram and decode it whole.
    ///
    /// Errors with [`ConnectionError::Codec`], carrying
    /// `CodecError::MalformedObject`, on a datagram with the PROPERTIES bit set
    /// and a Properties Length of 0, which
    /// [`Connection::close_for_data_stream`] closes the session over with
    /// MALFORMED_OBJECT.
    ///
    /// Errors with [`ConnectionError::PropertiesOnNonNormalStatus`] on a
    /// draft-23 datagram carrying properties on a status other than Normal.
    /// Draft-23 Section 11.2.1 builds the datagram's Properties field out of
    /// the structure defined in Section 11.1.2, and that section answers the
    /// combination with a session close — the same rule the subgroup form
    /// obeys, on the same receive side.
    ///
    /// A payload after a status is returned as
    /// [`ConnectionError::PayloadOnStatusDatagram`], and the session stays open.
    /// The codec's `DatagramHeader::decode_object` then reads the whole
    /// datagram. A Prior Group ID Gap or Prior Object ID Gap fault, and a
    /// property block that does not parse, are returned as
    /// [`ConnectionError::MalformedTrackDatagram`], for
    /// [`Connection::requests_to_cancel`]; its other refusals as
    /// [`ConnectionError::Codec`].
    ///
    /// The event is emitted before the checks, so an observer of the client's
    /// event stream still sees the datagram that caused the error.
    pub async fn recv_datagram(&self) -> Result<(AnyDatagramHeader, Bytes), ConnectionError> {
        let data = self.transport.recv_datagram().await?;
        let mut cursor = &data[..];
        let header = AnyDatagramHeader::decode(self.draft, &mut cursor)?;
        let consumed = data.len() - cursor.len();
        let payload = data.slice(consumed..);
        self.emit(ClientEvent::DatagramReceived {
            direction: Direction::Receive,
            header: header.clone(),
            payload_len: payload.len(),
        });
        // Refutable only in a build with more than one draft enabled;
        // in a single-draft build `AnyDatagramHeader` has one variant.
        #[allow(irrefutable_let_patterns)]
        if let AnyDatagramHeader::Draft23(h) = &header {
            // draft-23 Section 11.2.1: "If an endpoint receives a datagram with
            // the PROPERTIES bit set and an Properties Length of 0, it MUST close
            // the session with MALFORMED_OBJECT."
            if !h.properties_block_well_formed() {
                return Err(ConnectionError::Codec(CodecError::MalformedObject {
                    detail: "a datagram with the PROPERTIES bit set has a Properties Length of 0",
                }));
            }
            // draft-23 Section 11.2.1: "If an Object Datagram includes both the
            // STATUS bit and PROPERTIES bit, and the Object Status is not Normal
            // (0x0), the endpoint MUST close the session with
            // INVALID_OBJECT_STATUS, because only Normal Objects can have
            // Properties."
            if !h.properties_permitted() {
                return Err(ConnectionError::PropertiesOnNonNormalStatus {
                    object_id: h.object_id.into_inner(),
                    properties_len: h.properties.len(),
                    status: h.status(),
                });
            }
            if !h.permits_payload() && !payload.is_empty() {
                return Err(ConnectionError::PayloadOnStatusDatagram {
                    object_id: h.object_id.into_inner(),
                    payload_len: payload.len(),
                    status: h.object_status,
                });
            }
            // The codec's whole-datagram read, for the rules a header alone
            // does not show. A gap fault is returned with the datagram's alias,
            // for `requests_to_cancel`.
            let whole =
                moqtap_codec::draft23::data_stream::DatagramHeader::decode_object(&mut &data[..]);
            match whole {
                Ok(_) => {}
                Err(
                    error @ (CodecError::MalformedTrack { .. }
                    | CodecError::MalformedObjectProperties { .. }),
                ) => {
                    return Err(ConnectionError::MalformedTrackDatagram {
                        track_alias: h.track_alias.into_inner(),
                        error,
                    });
                }
                Err(error) => return Err(ConnectionError::Codec(error)),
            }
        }
        // A datagram is a whole object, so the connection can measure it
        // without help from the caller. It cannot *answer* the condition,
        // though: the answer is a reset of a request stream the caller holds,
        // so both data paths report and neither withdraws - see
        // `Connection::requests_to_cancel`.
        let meta = header.meta();
        // Refutable only in a build with more than one draft enabled.
        #[allow(irrefutable_let_patterns)]
        let properties: &[u8] =
            if let AnyDatagramHeader::Draft23(h) = &header { &h.properties } else { &[] };
        self.endpoint.note_received_object(
            meta.track_alias,
            ObjectLocation { group: meta.group_id, object: meta.object_id },
            object_role(meta.status),
            properties,
        )?;
        Ok((header, payload))
    }

    // -- Accessors --------------------------------------------------

    /// Access the underlying endpoint state machine.
    pub fn endpoint(&self) -> &Endpoint {
        &self.endpoint
    }

    /// Mutable access to the endpoint state machine.
    pub fn endpoint_mut(&mut self) -> &mut Endpoint {
        &mut self.endpoint
    }

    /// The SETUP message the server answered the handshake with.
    ///
    /// [`AnyControlMessage::fields`] renders it under
    /// this draft's own parameter names, in the order they arrived.
    pub fn server_setup(&self) -> &AnyControlMessage {
        &self.server_setup
    }

    /// The framed wire bytes of [`Self::server_setup`], as they arrived.
    ///
    /// Kept beside the decoded form because the encoding is evidence the
    /// decoding discards: two relays sending the same parameter can still
    /// disagree on how wide a varint they wrote it in.
    pub fn server_setup_raw(&self) -> Option<&[u8]> {
        self.server_setup_raw.as_deref()
    }

    /// Returns the draft version this connection is using.
    pub fn draft(&self) -> DraftVersion {
        self.draft
    }

    /// Close the session on the wire when the endpoint says a violation is
    /// fatal to it, and hand the error back unchanged.
    ///
    /// [`EndpointError::session_error_code`] answers `Some` for exactly the
    /// errors draft-23 tells the receiver to close the session over, and the
    /// endpoint has already moved its own state machine to Closed by the time
    /// this runs. Without this step that move is purely internal: the local
    /// endpoint refuses to start anything new while the peer, which is the one
    /// that broke the rule, sees a session that is still open and goes on
    /// sending. "MUST close the session" is a statement about the wire, so it takes a
    /// CONNECTION_CLOSE to satisfy it.
    ///
    /// The reason phrase is the error's own `Display` text, which names the
    /// message and the rule rather than repeating the numeric code the close
    /// already carries.
    ///
    /// Errors that answer `None` are recoverable and nothing is sent.
    fn close_for(&self, err: &EndpointError) {
        if let Some(code) = err.session_error_code() {
            // QUIC application error codes are 62-bit; every code in this
            // registry is far below `u32::MAX`, and saturating rather than
            // truncating means a future code that is not could never be
            // reported as a different, assigned one.
            let wire_code = u32::try_from(code.as_u64()).unwrap_or(u32::MAX);
            self.close(wire_code, err.to_string().as_bytes());
        }
    }

    /// [`close_for`](Self::close_for), then the error unchanged, for the
    /// common case where the endpoint's error is also what the caller returns.
    fn close_if_session_fatal(&self, err: EndpointError) -> ConnectionError {
        self.close_for(&err);
        ConnectionError::Endpoint(err)
    }

    /// Which of this draft's *own* `ConnectionError` variants this error is,
    /// and which kind of thing it says.
    ///
    /// The ten every draft carries answer `None` here: [`AnyConnectionError`]
    /// classifies those itself, once, and never asks a draft about them. What
    /// is left splits two ways, and the split is the reason this function
    /// exists — before it, both halves reached a caller as a sentence and read
    /// exactly alike. A [`LocalRefusal`] is this endpoint declining to write
    /// something, so nothing reached the wire and no relay is implicated; a
    /// [`PeerViolation`] is a peer having done something draft-23 forbids, and
    /// carries the session error code draft-23's own text answers it with.
    ///
    /// Matched exhaustively, with no wildcard arm and deliberately so: a
    /// variant added to this draft's error type has to arrive here as a compile
    /// error, beside the doc comment quoting the sentence it enforces, rather
    /// than as a silent [`ErrorCause::Unclassified`] in the facade.
    ///
    /// [`AnyConnectionError`]: crate::dispatch::AnyConnectionError
    /// [`ErrorCause::Unclassified`]: crate::dispatch::ErrorCause::Unclassified
    /// [`LocalRefusal`]: crate::above_codec_rules::DraftSpecificCause::LocalRefusal
    /// [`PeerViolation`]: crate::above_codec_rules::DraftSpecificCause::PeerViolation
    pub fn draft_specific_cause(
        err: &ConnectionError,
    ) -> Option<crate::above_codec_rules::DraftSpecificCause> {
        use crate::above_codec_rules::{AboveCodecRule, DraftSpecificCause};
        use moqtap_codec::draft23::error_codes::SessionErrorCode;

        match err {
            ConnectionError::Endpoint(_)
            | ConnectionError::Codec(_)
            | ConnectionError::Transport(_)
            | ConnectionError::VarInt(_)
            | ConnectionError::NoControlStream
            | ConnectionError::UnexpectedEnd
            | ConnectionError::StreamFinished
            | ConnectionError::InvalidAddress(_)
            | ConnectionError::TlsConfig(_)
            | ConnectionError::DataStreamState(_) => None,
            // This build decoding a message and then failing to narrow it to
            // its own draft. Nothing reached the wire and no peer is
            // implicated, which is the whole reason it is not
            // `ConnectionError::Codec`: under that name it would carry
            // a close code out of `codec_session_error_code` and
            // publish a relay for this build's defect. See the variant's own
            // doc.
            ConnectionError::ControlMessageNarrowing => {
                Some(crate::above_codec_rules::DraftSpecificCause::LocalRefusal)
            }
            // Section 11.1.2 states the rule and names the code in the same
            // sentence. The codec refuses such an Object only as a bare
            // `InvalidField`, so this layer is the one that names the rule,
            // and `close_for_data_stream` performs the close by
            // reading this same answer.
            ConnectionError::PropertiesOnNonNormalStatus { .. } => {
                Some(DraftSpecificCause::PeerViolation {
                    rule: AboveCodecRule::PropertiesOnNonNormalStatus,
                    close: Some(SessionErrorCode::InvalidObjectStatus.as_u64()),
                })
            }
            // Section 11.3: "the receiver SHOULD close the session with
            // TRUNCATED_OBJECT". `close_for_data_stream` reads this answer.
            ConnectionError::TruncatedObject => Some(DraftSpecificCause::PeerViolation {
                rule: AboveCodecRule::TruncatedObject,
                close: Some(SessionErrorCode::TruncatedObject.as_u64()),
            }),
            // The codec's refusal, carried with the alias its error does
            // not name. draft-23 Section 12.1 answers a Malformed Track by
            // cancelling requests, and `codec_session_error_code` names no
            // close for it.
            ConnectionError::MalformedTrackDatagram { error, .. } => {
                Some(DraftSpecificCause::Codec {
                    close: Self::codec_session_error_code(error).map(|code| code.as_u64()),
                    error: error.clone(),
                })
            }
            // Section 11.1.1 states this one as a property of a conforming
            // Object rather than as one of the cases a draft answers with a
            // close. The datagram is refused and the session is left running.
            // `close: None` is that reading, and it is what keeps a relay from
            // being published for a rule its draft attaches no consequence to.
            ConnectionError::PayloadOnStatusDatagram { .. } => {
                Some(DraftSpecificCause::PeerViolation {
                    rule: AboveCodecRule::PayloadOnStatusDatagram,
                    close: None,
                })
            }
            // Section 6.3: "Bidirectional streams MUST NOT begin with any
            // other message type unless negotiated. If they do, the peer MUST
            // close the Session with UNEXPECTED_MESSAGE." The session has
            // already been closed on the wire by the time this is returned, so
            // the code is carried here for a caller to read which rule was
            // answered, not for it to answer one again.
            ConnectionError::NonRequestOnRequestStream(_) => {
                Some(DraftSpecificCause::PeerViolation {
                    rule: AboveCodecRule::BidiStreamOpener,
                    close: Some(SessionErrorCode::UnexpectedMessage.as_u64()),
                })
            }
            // Three ways of handing this endpoint a message it will not write,
            // and one answer: nothing reached the wire, so nothing here is
            // evidence about a peer. Two are messages put on the control stream
            // that belong on a request stream of their own; the third is a
            // `respond_*` helper pointed at a request this endpoint opened,
            // which only the endpoint a request was opened *toward* may answer.
            ConnectionError::RequestOnControlStream(_)
            | ConnectionError::RequestStreamMessageOnControlStream(_)
            | ConnectionError::RespondedToOwnRequest(_) => Some(DraftSpecificCause::LocalRefusal),
            // Section 6.4.2.2 forbids a FIN before the response a request type
            // requires, and before PUBLISH_DONE on an established
            // subscription. Both are raised when *this* endpoint asks for the
            // FIN, so both are refusals rather than findings: the draft's rule
            // is being kept, not broken.
            ConnectionError::FinBeforeResponse(_) | ConnectionError::FinBeforePublishDone(_) => {
                Some(DraftSpecificCause::LocalRefusal)
            }
        }
    }

    /// The code to close the session with when a control message or a data
    /// stream could not be decoded because the peer broke a rule draft-23
    /// answers with a close.
    ///
    /// Draft-23 Section 12.2 gives each category its own code, and the
    /// sentence stating each rule names the one it takes. Where more than one
    /// applies, "an endpoint uses DELTA_ENCODING_OVERFLOW or
    /// FIELD_LENGTH_EXCEEDED in preference to the other codes."
    ///
    /// `None` for everything else, including [`CodecError::InvalidField`].
    /// That variant is shared by many unrelated malformations, only some of
    /// which the draft answers with a close, so treating it as fatal would
    /// close sessions the draft does not ask to be closed.
    pub fn codec_session_error_code(
        err: &CodecError,
    ) -> Option<moqtap_codec::draft23::error_codes::SessionErrorCode> {
        use moqtap_codec::draft23::error_codes::SessionErrorCode;
        use moqtap_codec::kvp::KvpError;
        match err {
            // Section 9: "If the length does not match the length of the
            // Message Body, the receiver MUST close the session with
            // MALFORMED_MESSAGE."
            CodecError::ControlMessageLengthMismatch { .. }
            // Section 9.4.2.1: "An endpoint that receives a REQUEST_ERROR with
            // Error Code REDIRECT and no REDIRECT Error Detail MUST close the
            // session with MALFORMED_MESSAGE."
            | CodecError::MissingErrorDetail { .. }
            // This draft's FETCH has no Fetch Type (Section 9.11), so its
            // decoder cannot produce this. A discriminator a reader cannot name
            // is a message it cannot parse, which Section 12.2 defines
            // MALFORMED_MESSAGE as: "A control message cannot be parsed or is
            // missing a required element".
            | CodecError::InvalidFetchType(_)
            // A LOCATION_FILTER (Section 9.19.9) or FILL_PARAMETERS (Section
            // 9.19.15) value whose fields run past the end of the message.
            // Neither carries a length of its own, so the Message Body is
            // shorter than its fields, and Section 9: "If the length does not
            // match the length of the Message Body, the receiver MUST close
            // the session with MALFORMED_MESSAGE."
            | CodecError::SubscriptionFilterMalformed { .. } => {
                Some(SessionErrorCode::MalformedMessage)
            }
            // Section 9: "An endpoint that receives an unknown message type
            // MUST close the session with UNEXPECTED_MESSAGE."
            CodecError::UnknownMessageType(_) => Some(SessionErrorCode::UnexpectedMessage),
            // Section 9: "If an endpoint receives a control message with a
            // length larger than its MAX_CONTROL_MESSAGE_SIZE, it MUST close
            // the session with FIELD_LENGTH_EXCEEDED."
            CodecError::ControlMessageExceedsLimit { .. }
            // Section 8.3: "The maximum length of a value is 2^16-1 bytes. If
            // an endpoint receives a length larger than the maximum, it MUST
            // close the session with FIELD_LENGTH_EXCEEDED."
            | CodecError::Kvp(KvpError::ValueTooLong(_))
            // Section 8.7, the 4096-byte Full Track Name maximum.
            | CodecError::TrackNameTooLong
            // Section 8.5, the 1024-byte Reason Phrase maximum.
            | CodecError::ReasonPhraseTooLong
            // Section 9.2, the 8192-byte New Session URI maximum.
            | CodecError::GoAwayUriTooLong => Some(SessionErrorCode::FieldLengthExceeded),
            // Section 8.7: "If an endpoint receives a Track Namespace
            // consisting of greater than 32 Track Namespace Fields, it MUST
            // close the session with INVALID_TRACK_NAMESPACE", and the same
            // code for a Track Namespace Field Length of 0.
            CodecError::InvalidNamespaceTupleSize(_) | CodecError::EmptyNamespaceField => {
                Some(SessionErrorCode::InvalidTrackNamespace)
            }
            // Section 2.4.1: "If an endpoint receives a Track Name with a length
            // of 0, it MUST close the session with a PROTOCOL_VIOLATION."
            CodecError::EmptyTrackName => Some(SessionErrorCode::ProtocolViolation),
            // The delta encodings: a Key-Value-Pair or Message Parameter Type
            // (Section 8.3 and Section 9.19), a Location Filter's end Group
            // (Section 9.19.9), a subgroup Object ID (Section 11.3.1), and a
            // fetch stream's Group and Object IDs (Section 11.4.1.1 and
            // Section 11.4.2.1). Each "MUST close the session with
            // DELTA_ENCODING_OVERFLOW".
            CodecError::KeyDeltaOverflow(..)
            | CodecError::FilterEndGroupOverflow { .. }
            | CodecError::ObjectIdOverflow(..)
            | CodecError::DeltaDecodingOverflow { .. } => {
                Some(SessionErrorCode::DeltaEncodingOverflow)
            }
            // Section 9.19: "An endpoint that receives an unknown Message
            // Parameter MUST close the session with INVALID_PARAMETER." Section
            // 9.19.1 closes with the same code over a parameter in a message
            // its definition does not name, the duplicate rule of Section 9.19
            // does, and so do the value ranges of GROUP_ORDER (Section 9.19.8),
            // LOCATION_FILTER's type (Section 9.19.9), PAUSED (Section 9.19.18)
            // and INCLUDE_PROPERTIES (Section 9.19.22).
            //
            // A repeated known Setup Option reaches `DuplicateParameter` too.
            // Section 9.1 says "Senders MUST NOT repeat the same Option Type in
            // a message unless the option definition explicitly allows multiple
            // instances" and names no receiver action; it takes the code the
            // codec's one duplicate variant carries.
            CodecError::UnknownMessageParameter(_)
            | CodecError::ParameterOutOfScope { .. }
            | CodecError::DuplicateParameter(_)
            | CodecError::ParameterValueOutOfRange { .. }
            | CodecError::InvalidFilterType(_) => Some(SessionErrorCode::InvalidParameter),
            // Section 8.3: "If a receiver understands a Type, and the following
            // Value or Length/Value does not match the serialization defined by
            // that Type, the receiver MUST close the session with
            // MALFORMED_KEY_VALUE." A Setup Option, a Property, or an Error
            // Detail. Section 10.4, Section 10.5 and Section 10.6 state the
            // same code for the Track Property value ranges.
            CodecError::KeyValueFormatting { .. }
            | CodecError::MalformedErrorDetail { .. }
            | CodecError::TrackPropertyValueOutOfRange { .. } => {
                Some(SessionErrorCode::MalformedKeyValue)
            }
            // Section 8.9: "If the Token structure cannot be decoded, including
            // an unknown Alias Type, the receiver MUST close the Session with
            // MALFORMED_AUTH_TOKEN."
            CodecError::MalformedAuthToken { .. } => Some(SessionErrorCode::MalformedAuthToken),
            // A `Connection` is always the client, so the SETUP that carried
            // the token is the server's, and Section 9.1.4's ROLE_VIOLATION
            // ("If a server receives Alias Type DELETE (0x0) or USE_ALIAS (0x2)
            // in a SETUP message") is not this endpoint's to send. No Alias is
            // registered before SETUP, and Section 8.9: "The receiver of a
            // message referencing an Alias that is not currently registered
            // MUST reject the message with UNKNOWN_AUTH_TOKEN_ALIAS."
            CodecError::TokenAliasInSetup { .. } => Some(SessionErrorCode::UnknownAuthTokenAlias),
            // Section 9.3: "If an endpoint receives Track Properties in one of
            // these messages it MUST close the session with INVALID_FIELD."
            CodecError::TrackPropertiesNotPermitted => Some(SessionErrorCode::InvalidField),
            // Section 6.4.1 for an unknown stream type and Section 11 for an
            // unknown datagram type, and Section 11.3.1 and Section 11.2.1 for
            // the Type values they list as invalid: each "MUST close the
            // session with INVALID_STREAM_HEADER".
            CodecError::UnknownStreamType(_)
            | CodecError::UnknownDatagramType(_)
            | CodecError::InvalidStreamTypeValue { .. }
            | CodecError::InvalidDatagramTypeValue { .. } => {
                Some(SessionErrorCode::InvalidStreamHeader)
            }
            // Section 11.4.1, Section 11.4.1.1 and Section 11.4.2.2: an
            // undefined Serialization Flags value above 127, and an entry that
            // refers to a prior Object there is none of, each "MUST close the
            // session with MALFORMED_OBJECT".
            CodecError::MalformedObject { .. } => Some(SessionErrorCode::MalformedObject),
            // Section 11.1.1: "If an endpoint receives any other value, it
            // SHOULD close the session with INVALID_OBJECT_STATUS (Section
            // 12.2)."
            CodecError::UnknownObjectStatus(_) => Some(SessionErrorCode::InvalidObjectStatus),
            // Everything this draft does not answer with a close, named rather
            // than swept up by a wildcard, so a new `CodecError` variant does
            // not compile until it has been placed.
            //
            // A Malformed Track (Section 12.1) costs the track and not the
            // session: "When a subscriber detects a Malformed Track, it MUST
            // cancel any corresponding subscription or fetches for that Track
            // from that publisher".
            CodecError::MalformedTrack { .. }
            | CodecError::MalformedObjectProperties { .. }
            // Section 11.1.1 states the payload rule as a MUST on the sender and
            // names no receiver action.
            | CodecError::PayloadNotPermitted { .. }
            | CodecError::InvalidForward(_)
            | CodecError::InvalidContentExists(_)
            | CodecError::UnexpectedEnd
            | CodecError::MessageTooLong(_)
            | CodecError::VarInt(_)
            | CodecError::InvalidField
            | CodecError::InvalidRange(..)
            | CodecError::ParameterLengthMismatch(_)
            | CodecError::EndOfTrackObjectId(_)
            | CodecError::ParametersOutOfOrder(..)
            | CodecError::ExtensionsOnNonExistentObject(_)
            | CodecError::InvalidRequiredRequestIdDelta(..)
            | CodecError::UnsupportedDraft(_)
            | CodecError::Kvp(
                KvpError::MissingLength | KvpError::UnexpectedEnd | KvpError::VarInt(_),
            ) => None,
        }
    }

    /// Close the session on the wire when a decode failure is one draft-23
    /// answers with a close, and hand the error back unchanged.
    ///
    /// The codec's counterpart to
    /// [`close_if_session_fatal`](Self::close_if_session_fatal). Without it
    /// every bound the decoder enforces would stop at *this endpoint refused the
    /// frame* while the peer, which is the one that broke the rule, saw a
    /// session that was still open and went on sending. "MUST close the session" is a statement
    /// about the wire.
    fn close_for_codec(&self, err: ConnectionError) -> ConnectionError {
        if let ConnectionError::Codec(inner) = &err {
            if let Some(code) = Self::codec_session_error_code(inner) {
                // QUIC application error codes are 62-bit; every code in this
                // registry is far below `u32::MAX`, and saturating rather than
                // truncating means a future code that is not could never be
                // reported as a different, assigned one.
                let wire_code = u32::try_from(code.as_u64()).unwrap_or(u32::MAX);
                self.close(wire_code, inner.to_string().as_bytes());
            }
        }
        err
    }

    /// Name every request whose stream the caller must reset, for a track a
    /// data path has just found malformed.
    ///
    /// Section 12.1 answers its whole list of conditions at once: "it MUST
    /// cancel any corresponding subscription or fetches for that Track from
    /// that publisher". On this draft cancelling a request is a transport
    /// operation rather than a message — Section 6.4.2.3: "Implementations cancel a request by
    /// abruptly terminating any directions of the stream that are still open,
    /// using RESET_STREAM for a direction they are sending and STOP_SENDING for
    /// a direction they are receiving."
    ///
    /// # Why this returns ids instead of doing it
    ///
    /// Because the streams are the caller's. Every request on this draft lives
    /// at the front of a bidirectional stream of its own, and
    /// [`Connection::recv_on_request_stream`] hands that stream back as a
    /// [`RequestStream`]. There is no handle here to reset. So the connection
    /// does the half it can — note the track, and work out which requests
    /// receive it — and the caller passes each id to
    /// [`Connection::cancel_request_stream`], which resets the stream *and*
    /// moves the endpoint's record.
    ///
    /// # Every data path comes here
    ///
    /// Subgroup streams, FETCH streams and datagrams report the condition the
    /// same way, so there is one entry point. Pass it whatever error a read
    /// returned; anything that is not this condition gives back an empty list.
    ///
    /// Empty is not "the track was fine" — it is also what an alias no live
    /// binding names gives, and what a track this endpoint only publishes
    /// gives.
    pub fn requests_to_cancel(&self, err: &ConnectionError) -> Vec<VarInt> {
        match err {
            ConnectionError::Endpoint(EndpointError::ObjectPastFinalObject { alias, .. }) => {
                self.endpoint.requests_for_malformed_track(
                    *alias,
                    MalformedTrackCondition::ObjectPastFinalObject,
                )
            }
            ConnectionError::Endpoint(EndpointError::ObjectExistenceContradicted {
                namespace,
                name,
                ..
            }) => self.endpoint.requests_for_malformed_track_named(
                namespace,
                name,
                MalformedTrackCondition::ExistenceContradicted,
            ),
            ConnectionError::MalformedTrackDatagram { track_alias, error } => {
                malformed_track_condition(error).map_or_else(Vec::new, |condition| {
                    self.endpoint.requests_for_malformed_track(*track_alias, condition)
                })
            }
            _ => Vec::new(),
        }
    }

    /// [`Connection::requests_to_cancel`] for an error read from the subgroup
    /// stream `header` opened, which also names the requests to cancel over
    /// the [`CodecError::MalformedTrack`] that
    /// [`FramedRecvStream::read_subgroup_object`] returns for a Prior Group ID
    /// Gap or Prior Object ID Gap fault (draft-23 Section 10.8,
    /// draft-23 Section 10.9), and over the
    /// [`CodecError::MalformedObjectProperties`] it returns for a property block
    /// that does not parse (draft-23 Section 10.7). Those errors name no track, so
    /// the header supplies its alias.
    ///
    /// draft-23 Section 12.1: "When a subscriber detects a Malformed Track, it
    /// MUST cancel any corresponding subscription or fetches for that Track
    /// from that publisher (see Section 6.4.2.3)". The session stays open.
    pub fn requests_to_cancel_for_subgroup_stream(
        &self,
        header: &AnySubgroupHeader,
        err: &ConnectionError,
    ) -> Vec<VarInt> {
        if let ConnectionError::Codec(codec) = err {
            if let Some(condition) = malformed_track_condition(codec) {
                return self.endpoint.requests_for_malformed_track(header.track_alias(), condition);
            }
        }
        self.requests_to_cancel(err)
    }

    /// [`Connection::requests_to_cancel`] for an error read from the fetch stream
    /// `header` opened, which also names the requests to cancel over the
    /// [`CodecError::MalformedTrack`] that
    /// [`FramedRecvStream::read_fetch_object`] returns for a Prior Group ID
    /// Gap or Prior Object ID Gap fault (draft-23 Section 10.8,
    /// draft-23 Section 10.9), and over the
    /// [`CodecError::MalformedObjectProperties`] it returns for a property block
    /// that does not parse (draft-23 Section 10.7). Those errors name no track, so
    /// the track the FETCH under the header's Request ID asked for is the one.
    ///
    /// draft-23 Section 12.1: "When a subscriber detects a Malformed Track, it
    /// MUST cancel any corresponding subscription or fetches for that Track
    /// from that publisher (see Section 6.4.2.3)". The session stays open.
    pub fn requests_to_cancel_for_fetch_stream(
        &self,
        header: &AnyFetchHeader,
        err: &ConnectionError,
    ) -> Vec<VarInt> {
        if let ConnectionError::Codec(codec) = err {
            if let Some(condition) = malformed_track_condition(codec) {
                return self
                    .endpoint
                    .requests_for_malformed_fetch_track(header.request_id(), condition);
            }
        }
        self.requests_to_cancel(err)
    }

    /// Close the session when a failure raised while reading a *data* stream is
    /// one draft-23 answers with a close. Reports whether it closed.
    ///
    /// [`recv_control`](Self::recv_control) does this for itself, because it
    /// owns both the stream and the connection. A data stream does not:
    /// [`accept_subgroup_stream`](Self::accept_subgroup_stream) hands the caller
    /// a [`FramedRecvStream`], which holds no connection and so cannot close
    /// one, and the reads that raise these failures happen there. The caller is
    /// the only party holding both halves, which is what this is for:
    ///
    /// ```no_run
    /// # async fn f(conn: &moqtap_client::draft23::connection::Connection,
    /// #            framed: &mut moqtap_client::draft23::connection::FramedRecvStream)
    /// # -> Result<(), Box<dyn std::error::Error>> {
    /// if let Err(err) = framed.read_subgroup_object().await {
    ///     conn.close_for_data_stream(&err);
    ///     return Err(err.into());
    /// }
    /// # Ok(()) }
    /// ```
    ///
    /// Splitting it this way rather than closing inside the reader keeps a
    /// caller that is deliberately permissive — a tool reproducing a capture,
    /// say — able to read a violating stream and report it without tearing the
    /// session down. The rule is stated at endpoints, and this is where an
    /// endpoint decides it is one.
    ///
    /// Only [`ConnectionError::Codec`] failures are matched, against the same
    /// `codec_session_error_code` table the
    /// control path uses, so a rule is answered with one code whichever stream
    /// carried it. The Object ID delta wrap of Section 11.3.1 is the entry that
    /// can only arrive this way.
    pub fn close_for_data_stream(&self, err: &ConnectionError) -> bool {
        use crate::above_codec_rules::DraftSpecificCause;

        match err {
            ConnectionError::Codec(inner) => {
                let Some(code) = Self::codec_session_error_code(inner) else { return false };
                let wire_code = u32::try_from(code.as_u64()).unwrap_or(u32::MAX);
                self.close(wire_code, inner.to_string().as_bytes());
                true
            }
            // Not a `Codec` failure: the codec refuses such an Object as a bare
            // `InvalidField`, which `read_subgroup_object` narrows to this
            // crate's own variant, so the mapping table above never sees it.
            //
            // The code comes from `draft_specific_cause` rather than from a
            // constant here, so this draft's reading of its own sentence is
            // written down once and a caller who reads the error as a value
            // sees the same code the peer was sent.
            ConnectionError::PropertiesOnNonNormalStatus { .. }
            | ConnectionError::TruncatedObject => {
                let Some(DraftSpecificCause::PeerViolation { close: Some(code), .. }) =
                    Self::draft_specific_cause(err)
                else {
                    return false;
                };
                // Saturate rather than truncate, so a future code above
                // `u32::MAX` is never reported as a different assigned one.
                self.close(u32::try_from(code).unwrap_or(u32::MAX), err.to_string().as_bytes());
                true
            }
            _ => false,
        }
    }

    /// Close the connection.
    pub fn close(&self, code: u32, reason: &[u8]) {
        self.emit(ClientEvent::Closed { code, reason: reason.to_vec() });
        self.transport.close(code, reason);
    }
}

/// The Malformed Track condition a codec refusal on a data path is, on this
/// draft, or `None` for a refusal that is not one.
fn malformed_track_condition(err: &CodecError) -> Option<MalformedTrackCondition> {
    match err {
        CodecError::MalformedTrack { .. } => Some(MalformedTrackCondition::PriorGap),
        // draft-23 Section 10.7: "A Key-Value-Pair cannot be parsed."
        CodecError::MalformedObjectProperties { .. } => {
            Some(MalformedTrackCondition::MalformedPropertyBlock)
        }
        _ => None,
    }
}

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

    /// Only a refusal of what the peer sent is reported as a refused frame,
    /// and only with the bytes it was about.
    #[test]
    fn only_a_refusal_of_the_peers_frame_is_reported() {
        let raw = Some(vec![0x01]);
        let decoder = ConnectionError::Codec(CodecError::UnexpectedEnd);
        assert!(Connection::refused_event(&decoder, None, raw.clone()).is_some());
        assert!(Connection::refused_event(&decoder, None, None).is_none());
        let local = ConnectionError::Endpoint(EndpointError::NotActive);
        assert!(Connection::refused_event(&local, None, raw).is_none());
    }

    /// This build failing to narrow a message it decoded is never a finding
    /// about the peer.
    ///
    /// The arm that raises `ControlMessageNarrowing` is unreachable — this
    /// draft's decoder can only hand back this draft's variant — and nothing
    /// pins that. What is pinned here is the half that matters.
    /// `CodecError::UnknownMessageType(0)` is what the arm must not raise:
    /// `codec_session_error_code` answers it with a session close on every draft in range, so the
    /// day the narrowing failed a conformance
    /// probe would publish a relay for sending a control message type this
    /// draft does not assign — with `0x00` attached as the codepoint that
    /// proved it, which is an accusation better evidenced than any real one
    /// this build makes. The section stating that rule is numbered differently
    /// on every draft, and the point does not turn on the number.
    #[test]
    fn a_message_this_build_could_not_narrow_names_nobody() {
        use crate::dispatch::{AnyConnectionError, ErrorCause};

        let err: AnyConnectionError = ConnectionError::ControlMessageNarrowing.into();
        assert!(err.is_local(), "a narrowing this build could not do is this build's");
        assert_eq!(
            err.cause(),
            &ErrorCause::Facade,
            "nothing reached the wire, so there is no rule and no close code to read"
        );
    }

    /// Draft-23 uses MoQT's variable-length integer, whose length is the
    /// number of leading 1 bits in the first byte, not RFC 9000's two-bit
    /// prefix. Control framing measures the type field with it before any
    /// bytes past the first have arrived.
    #[test]
    fn varint_len_follows_the_moqt_encoding() {
        let draft = DraftVersion::Draft23;
        assert_eq!(draft.varint_len(0x00), 1);
        assert_eq!(draft.varint_len(0x7F), 1);
        assert_eq!(draft.varint_len(0x80), 2);
        assert_eq!(draft.varint_len(0xBF), 2);
        assert_eq!(draft.varint_len(0xC0), 3);
        assert_eq!(draft.varint_len(0xFF), 9);
        // SETUP's type id, 0x2F00, is two bytes here and four under RFC 9000.
        assert_eq!(draft.varint_len(0xAF), 2);
    }

    #[test]
    fn client_config_alpn_quic_draft23() {
        let config = ClientConfig {
            draft: DraftVersion::Draft23,
            transport: TransportType::Quic,
            skip_cert_verification: false,
            ca_certs: Vec::new(),
            setup_parameters: Vec::new(),
        };
        assert_eq!(config.alpn(), vec![b"moqt-23".to_vec()]);
    }

    #[test]
    fn client_config_alpn_webtransport() {
        let config = ClientConfig {
            draft: DraftVersion::Draft23,
            transport: TransportType::WebTransport { url: "https://example.com".to_string() },
            skip_cert_verification: false,
            ca_certs: Vec::new(),
            setup_parameters: Vec::new(),
        };
        assert_eq!(config.alpn(), vec![b"h3".to_vec()]);
    }

    /// `MOQT_ALPN` is the ALPN a client configured for this draft offers.
    #[test]
    fn moqt_alpn_is_the_one_a_client_offers() {
        // Checked against what a client configured for this draft puts on
        // the wire, and only then against the literal.
        let config = ClientConfig {
            draft: DraftVersion::Draft23,
            transport: TransportType::Quic,
            skip_cert_verification: false,
            ca_certs: Vec::new(),
            setup_parameters: Vec::new(),
        };
        assert_eq!(
            config.alpn(),
            vec![MOQT_ALPN.to_vec()],
            "MOQT_ALPN is {:?}; a draft-{} client offers {:?}",
            String::from_utf8_lossy(MOQT_ALPN),
            23,
            config
                .alpn()
                .iter()
                .map(|a| String::from_utf8_lossy(a).into_owned())
                .collect::<Vec<_>>(),
        );
        assert_eq!(MOQT_ALPN, b"moqt-23");
    }

    /// Draft-23 Section 6.3 names seven message types a bidirectional stream
    /// may begin with, and no others. The set is checked against the raw
    /// numbers this draft's registry assigns rather than against the names,
    /// so a renumbered variant is caught even though its spelling is the
    /// same, and so is a set that leaves SUBSCRIBE_TRACKS (0x51) out.
    ///
    /// Every type this draft assigns is classified: the loop walks the whole
    /// assigned range and asks the classifier about each one it finds.
    #[test]
    fn only_seven_message_types_open_a_request_stream() {
        // TRACK_STATUS, SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE,
        // SUBSCRIBE_NAMESPACE and SUBSCRIBE_TRACKS, written as the numbers
        // draft-23 assigns them.
        let mut expected = vec![0x0D, 0x03, 0x1D, 0x16, 0x06, 0x50, 0x51];
        expected.sort_unstable();

        let mut opens = Vec::new();
        for id in 0..=CONTROL_STREAM_TYPE {
            if let Some(ty) = MessageType::from_id(id) {
                if starts_a_request_stream(ty) {
                    opens.push(id);
                }
            }
        }
        opens.sort_unstable();

        assert_eq!(
            opens, expected,
            "the types that open a request stream are {opens:?}; \
             draft-23 Section 6.3 names {expected:?}"
        );
    }

    /// The kind a request helper labels its stream with must name the message
    /// that helper actually writes, and the check is made against the type
    /// varint the encoded message leads with — the byte a peer reads to
    /// decide whether the bidirectional stream is legal.
    ///
    /// SUBSCRIBE_NAMESPACE (0x50) and SUBSCRIBE_TRACKS (0x51) are adjacent,
    /// which makes them the pair most easily mislabelled.
    #[test]
    fn each_request_kind_labels_the_message_its_helper_writes() {
        use crate::draft23::endpoint::Endpoint;
        use moqtap_codec::draft23::message::Setup;

        let v = |n: u64| VarInt::from_u64(n).unwrap();
        let ns = TrackNamespace(vec![b"ns".to_vec()]);

        let mut ep = Endpoint::new(Role::Client);
        ep.connect().unwrap();
        let _ = ep.send_setup(vec![]).unwrap();
        ep.receive_setup(&Setup { options: vec![] }).unwrap();

        let (_, subscribe) = ep.subscribe(ns.clone(), b"t".to_vec(), vec![]).unwrap();
        let built = vec![
            (RequestKind::Subscribe, subscribe),
            // One FETCH: draft-23 Section 9.11 has no Fetch Type field, so there
            // is no second shape for this to label.
            (RequestKind::Fetch, ep.fetch(ns.clone(), b"t".to_vec(), vec![]).unwrap().1),
            (
                RequestKind::SubscribeNamespace,
                ep.subscribe_namespace(ns.clone(), vec![]).unwrap().1,
            ),
            (RequestKind::SubscribeTracks, ep.subscribe_tracks(ns.clone(), vec![]).unwrap().1),
            (RequestKind::PublishNamespace, ep.publish_namespace(ns.clone(), vec![]).unwrap().1),
            (
                RequestKind::TrackStatus,
                ep.track_status(ns.clone(), b"t".to_vec(), vec![]).unwrap().1,
            ),
            (
                RequestKind::Publish,
                ep.publish(ns.clone(), b"t".to_vec(), v(7), vec![], vec![]).unwrap().1,
            ),
        ];

        for (kind, msg) in built {
            let mut wire = Vec::new();
            msg.encode(&mut wire).unwrap();
            let mut cursor = &wire[..];
            let on_the_wire =
                DraftVersion::Draft23.decode_varint(&mut cursor).unwrap().into_inner();
            assert_eq!(
                kind.message_type().id(),
                on_the_wire,
                "{kind:?} is labelled {} but its message leads with {on_the_wire}",
                kind.message_type().id(),
            );
            assert!(
                starts_a_request_stream(kind.message_type()),
                "{kind:?} labels a message type that may not begin a bidirectional stream",
            );
        }
    }

    /// The classifier the accept path runs and the one
    /// [`Connection::send_control`] runs must answer alike for every message
    /// type this draft assigns, or a message could be refused on the control
    /// stream and refused again as the opening of a request stream — leaving
    /// no legal place for it.
    #[test]
    fn the_two_request_stream_classifiers_agree() {
        let mut classified = 0;
        for id in 0..=CONTROL_STREAM_TYPE {
            let Some(ty) = MessageType::from_id(id) else { continue };
            classified += 1;
            let kind = RequestKind::from_message_type(ty);
            assert_eq!(
                kind.is_some(),
                starts_a_request_stream(ty),
                "type {id} {} a request stream but from_message_type calls it {kind:?}",
                if starts_a_request_stream(ty) { "opens" } else { "does not open" },
            );
            if let Some(kind) = kind {
                assert_eq!(
                    kind.message_type(),
                    ty,
                    "from_message_type sent type {id} to {kind:?}, which names a different message",
                );
            }
        }
        assert!(classified > 7, "the loop found only {classified} assigned message types");
    }

    /// A stream this endpoint opened is cancelled when its handle is dropped;
    /// one the peer opened is reset as unserved. Both codes are on the wire,
    /// so they may not be the same number.
    #[test]
    fn the_two_abandonment_codes_are_distinct() {
        assert_eq!(REQUEST_CANCELLED, 0x1);
        assert_eq!(REQUEST_UNANSWERED, 0x0);
        assert_ne!(
            REQUEST_CANCELLED, REQUEST_UNANSWERED,
            "a peer cannot tell a rejected request from a dropped one if both reset with the same code",
        );
    }

    #[test]
    fn transport_type_debug() {
        let quic = TransportType::Quic;
        assert!(format!("{quic:?}").contains("Quic"));

        let wt = TransportType::WebTransport { url: "https://example.com".to_string() };
        assert!(format!("{wt:?}").contains("WebTransport"));
    }
}

#[cfg(test)]
mod accept_on_the_wire {
    //! The accept path against a real QUIC peer.
    //!
    //! Everything the responder side does that a peer can see — which stream a
    //! response goes out on, which code an abandoned request stream is reset
    //! with, whether a FIN arrives before the messages that direction still
    //! owes, and which session error code a violation closes with — is
    //! observable only from the other end of a connection. A test holding an
    //! endpoint alone is satisfied by state that never reached the wire, so
    //! these hold the other end.
    //!
    //! The peer is raw `quinn` for everything the accept path decides. It does
    //! use [`FramedRecvStream`] to decode what comes back, because control
    //! framing is shared by both directions and gated elsewhere; the
    //! classification, registration, routing and stream-closing under test here
    //! are not part of it.

    use super::*;
    use std::sync::Arc;

    use std::net::SocketAddr;
    use std::time::Duration;

    use moqtap_codec::draft23::message::{
        GoAway, Publish, PublishDone, Setup, Subscribe, SubscribeNamespace,
    };

    /// Session termination codes, draft-23 Section 16.12.1.
    const UNEXPECTED_MESSAGE: u64 = 0x303;
    const INVALID_REQUEST_ID: u64 = 0x4;
    const UNKNOWN_AUTH_TOKEN_ALIAS: u64 = 0x17;
    const TRUNCATED_OBJECT: u64 = 0x203;

    /// Long enough that a loaded machine cannot fail a test that would
    /// otherwise pass, short enough that a hang is reported rather than run to
    /// the harness timeout.
    const PATIENCE: Duration = Duration::from_secs(10);

    /// How long to wait before concluding that nothing more is coming. Used
    /// only where the absence of a message is the assertion, so it is a floor
    /// on the test's run time and is kept small.
    const QUIET: Duration = Duration::from_millis(300);

    fn v(n: u64) -> VarInt {
        VarInt::from_u64(n).unwrap()
    }

    fn ns() -> TrackNamespace {
        TrackNamespace(vec![b"live".to_vec()])
    }

    fn encode(msg: ControlMessage) -> Vec<u8> {
        let mut buf = Vec::new();
        AnyControlMessage::Draft23(msg).encode(&mut buf).expect("encode");
        buf
    }

    /// A SUBSCRIBE carrying `id`. Odd ids are the ones a server allocates,
    /// which is what a client's peer must use.
    fn subscribe(id: u64) -> ControlMessage {
        ControlMessage::Subscribe(Subscribe {
            request_id: v(id),
            track_namespace: ns(),
            track_name: b"video".to_vec(),
            parameters: vec![],
        })
    }

    fn request_update(id: u64) -> ControlMessage {
        ControlMessage::RequestUpdate(moqtap_codec::draft23::message::RequestUpdate {
            request_id: v(id),
            parameters: vec![],
        })
    }

    fn peer_fetch(id: u64) -> ControlMessage {
        ControlMessage::Fetch(moqtap_codec::draft23::message::Fetch {
            request_id: v(id),
            track_namespace: ns(),
            track_name: b"video".to_vec(),
            parameters: vec![],
        })
    }

    fn fetch_ok() -> RequestOk {
        RequestOk { parameters: vec![], track_properties: vec![] }
    }

    fn subscribe_ok() -> SubscribeOk {
        SubscribeOk { track_alias: v(7), parameters: vec![], track_properties: vec![] }
    }

    fn request_error() -> RequestError {
        RequestError {
            error_code: v(0x1),
            retry_interval: v(0),
            reason_phrase: b"no".to_vec(),
            error_details: vec![],
        }
    }

    fn init_crypto() {
        let _ = rustls::crypto::ring::default_provider().install_default();
    }

    /// A quinn server on a loopback port, offering this draft's ALPN.
    fn server_endpoint() -> (quinn::Endpoint, SocketAddr) {
        use rcgen::{CertificateParams, KeyPair, PKCS_ECDSA_P256_SHA256};
        use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};

        let key_pair = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).expect("keypair");
        let params = CertificateParams::new(vec!["localhost".into()]).expect("params");
        let cert = params.self_signed(&key_pair).expect("self-sign");
        let cert_der = CertificateDer::from(cert.der().to_vec());
        let key_der = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key_pair.serialize_der()));

        let mut server_crypto = rustls::ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(vec![cert_der], key_der)
            .expect("server cert");
        server_crypto.alpn_protocols = vec![DraftVersion::Draft23.quic_alpn().to_vec()];
        let server_crypto =
            quinn::crypto::rustls::QuicServerConfig::try_from(server_crypto).expect("quic crypto");
        let server_config = quinn::ServerConfig::with_crypto(Arc::new(server_crypto));
        let endpoint = quinn::Endpoint::server(server_config, "127.0.0.1:0".parse().unwrap())
            .expect("bind server");
        let addr = endpoint.local_addr().expect("local_addr");
        (endpoint, addr)
    }

    async fn connect_client(addr: SocketAddr) -> Result<Connection, ConnectionError> {
        Connection::connect(
            &addr.to_string(),
            ClientConfig {
                draft: DraftVersion::Draft23,
                transport: TransportType::Quic,
                skip_cert_verification: true,
                ca_certs: Vec::new(),
                setup_parameters: Vec::new(),
            },
        )
        .await
    }

    /// The peer's half of the setup exchange: read the client's SETUP off its
    /// unidirectional control stream, answer with one of our own.
    ///
    /// Both control streams are handed back so they stay open for the
    /// connection's life. Dropping a quinn receive stream sends STOP_SENDING
    /// and dropping a send stream resets it, either of which would look to the
    /// client like the control plane failing.
    async fn peer_handshake(
        endpoint: &quinn::Endpoint,
    ) -> (quinn::Connection, quinn::SendStream, quinn::RecvStream) {
        let conn = endpoint.accept().await.expect("accept").await.expect("tls handshake");
        let mut client_control = conn.accept_uni().await.expect("accept_uni");
        let mut seen = Vec::new();
        let mut chunk = [0u8; 1024];
        while seen.len() < 3 {
            match client_control.read(&mut chunk).await.expect("read SETUP") {
                Some(n) => seen.extend_from_slice(&chunk[..n]),
                None => break,
            }
        }
        assert!(!seen.is_empty(), "the client sent no SETUP");
        let mut ours = conn.open_uni().await.expect("open_uni");
        ours.write_all(&encode(ControlMessage::Setup(Setup { options: Vec::new() })))
            .await
            .expect("write SETUP");
        (conn, ours, client_control)
    }

    /// A connected client and the peer holding the other end.
    struct Loopback {
        conn: Connection,
        peer: quinn::Connection,
        _endpoint: quinn::Endpoint,
        _control_send: quinn::SendStream,
        _control_recv: quinn::RecvStream,
    }

    async fn loopback() -> Loopback {
        init_crypto();
        let (endpoint, addr) = server_endpoint();
        let (client, peer) = tokio::join!(connect_client(addr), peer_handshake(&endpoint));
        let (peer, control_send, control_recv) = peer;
        Loopback {
            conn: client.expect("client connect"),
            peer,
            _endpoint: endpoint,
            _control_send: control_send,
            _control_recv: control_recv,
        }
    }

    fn framed(recv: quinn::RecvStream) -> FramedRecvStream {
        FramedRecvStream::new(RecvStream::Quic(recv), DraftVersion::Draft23)
    }

    /// Read one control message the client wrote, failing rather than hanging.
    async fn next_control(recv: &mut FramedRecvStream) -> ControlMessage {
        let (any, _) = tokio::time::timeout(PATIENCE, recv.read_control(false))
            .await
            .expect("the client wrote nothing")
            .expect("read control");
        match any {
            AnyControlMessage::Draft23(msg) => msg,
            #[allow(unreachable_patterns)]
            other => panic!("expected a draft-23 message, got {other:?}"),
        }
    }

    async fn assert_closed_with(peer: &quinn::Connection, code: u64, phrase: &str) {
        let reason = tokio::time::timeout(PATIENCE, peer.closed())
            .await
            .expect("the client reported the violation but never closed the connection");
        match reason {
            quinn::ConnectionError::ApplicationClosed(frame) => {
                assert_eq!(u64::from(frame.error_code), code, "the close carried the wrong code");
                let text = String::from_utf8_lossy(&frame.reason).to_string();
                assert!(
                    text.contains(phrase),
                    "the close reason should name the rule that was broken; got {text:?}",
                );
            }
            other => panic!("expected an application close, got {other:?}"),
        }
    }

    /// draft-23 Section 11.3: "If a stream ends gracefully (i.e., the stream
    /// terminates with a FIN) in the middle of a serialized Object, the
    /// receiver SHOULD close the session with TRUNCATED_OBJECT." A FIN after a
    /// whole Object is the stream's end and closes nothing.
    #[tokio::test]
    async fn a_fin_inside_an_object_closes_with_truncated_object() {
        let lb = loopback().await;

        // `subgroup-single-object`, then a FIN.
        let mut whole = lb.peer.open_uni().await.expect("open_uni");
        whole
            .write_all(&[0x10, 0x01, 0x00, 0x80, 0x00, 0x04, 0xde, 0xad, 0xbe, 0xef])
            .await
            .unwrap();
        whole.finish().unwrap();
        let (_, mut stream) = lb.conn.accept_subgroup_stream().await.expect("subgroup stream");
        stream.read_subgroup_object().await.expect("the whole Object");
        let end = stream.read_subgroup_object().await.expect_err("the stream ended");
        assert!(matches!(end, ConnectionError::UnexpectedEnd), "{end}");
        assert!(!lb.conn.close_for_data_stream(&end));

        // The same Object cut off two bytes into its payload.
        let mut cut = lb.peer.open_uni().await.expect("open_uni");
        cut.write_all(&[0x10, 0x01, 0x00, 0x80, 0x00, 0x04, 0xde, 0xad]).await.unwrap();
        cut.finish().unwrap();
        let (_, mut stream) = lb.conn.accept_subgroup_stream().await.expect("subgroup stream");
        let end = stream.read_subgroup_object().await.expect_err("the stream ended");
        assert!(matches!(end, ConnectionError::TruncatedObject), "{end}");
        assert!(lb.conn.close_for_data_stream(&end));
        assert_closed_with(&lb.peer, TRUNCATED_OBJECT, "FIN in the middle of an Object").await;
    }

    /// draft-23 Section 9: "An endpoint MUST NOT send a control message with
    /// a length larger than the peer's MAX_CONTROL_MESSAGE_SIZE (Section
    /// 9.1.8)." The server advertises 70000. A SUBSCRIBE between 65535 and
    /// that is sent, and a longer one is refused before it is written.
    #[tokio::test]
    async fn a_message_is_sent_under_the_servers_limit() {
        use moqtap_codec::varint::Moqt18;

        init_crypto();
        let (endpoint, addr) = server_endpoint();
        let peer = async {
            let conn = endpoint.accept().await.expect("accept").await.expect("tls handshake");
            let mut value = Vec::new();
            VarInt::from_u64_moqt(70_000).encode_moqt::<Moqt18>(&mut value);
            // MAX_CONTROL_MESSAGE_SIZE (0x09): a Length, then one vi64.
            let mut body = vec![0x09, value.len() as u8];
            body.extend_from_slice(&value);
            let mut setup = vec![0xaf, 0x00, body.len() as u8];
            setup.extend_from_slice(&body);
            let mut ours = conn.open_uni().await.expect("open_uni");
            ours.write_all(&setup).await.expect("write SETUP");
            (conn, ours)
        };
        let (client, (_peer, _ours)) = tokio::join!(connect_client(addr), peer);
        let mut conn = client.expect("client connect");

        // Two AUTHORIZATION TOKENs, Alias Type USE_VALUE (0x3), Token Type 0.
        let tokens = |len: usize| {
            let mut token = vec![0x03, 0x00];
            token.extend(std::iter::repeat_n(b't', len));
            let token =
                KeyValuePair { key: v(0x03), value: moqtap_codec::kvp::KvpValue::Bytes(token) };
            vec![token.clone(), token]
        };
        if let Err(e) = conn.subscribe(ns(), b"video".to_vec(), tokens(33_000)).await {
            panic!("a SUBSCRIBE of about 66000 bytes was refused under a limit of 70000: {e}");
        }
        match conn.subscribe(ns(), b"video".to_vec(), tokens(36_000)).await {
            Err(ConnectionError::Codec(CodecError::ControlMessageExceedsLimit {
                limit: 70_000,
                ..
            })) => {}
            Err(other) => panic!("expected ControlMessageExceedsLimit, got {other}"),
            Ok(_) => panic!("a SUBSCRIBE longer than the server's limit was sent"),
        }
    }

    /// The server's SETUP names an Alias, which no SETUP can: none is
    /// registered yet. draft-23 Section 8.9: "The receiver of a message
    /// referencing an Alias that is not currently registered MUST reject the
    /// message with UNKNOWN_AUTH_TOKEN_ALIAS." The bytes are the corpus vector
    /// `with-auth-token-delete`.
    #[tokio::test]
    async fn a_server_setup_naming_an_alias_closes_with_unknown_auth_token_alias() {
        init_crypto();
        let (endpoint, addr) = server_endpoint();
        let peer = async {
            let conn = endpoint.accept().await.expect("accept").await.expect("tls handshake");
            let mut ours = conn.open_uni().await.expect("open_uni");
            ours.write_all(&[0xaf, 0x00, 0x04, 0x03, 0x02, 0x00, 0x00]).await.expect("write SETUP");
            (conn, ours)
        };
        let (client, (peer, _ours)) = tokio::join!(connect_client(addr), peer);
        match client {
            Err(ConnectionError::Codec(CodecError::TokenAliasInSetup { alias_type: 0 })) => {}
            Err(other) => panic!("expected TokenAliasInSetup, got {other}"),
            Ok(_) => panic!("the client accepted a SETUP that names an Alias"),
        }
        assert_closed_with(&peer, UNKNOWN_AUTH_TOKEN_ALIAS, "Alias Type").await;
    }

    /// A peer's SUBSCRIBE is accepted, answered on the stream it arrived on,
    /// and cannot be finished until each message draft-23 Section 6.4.2.2
    /// requires has been sent.
    #[tokio::test]
    async fn a_peers_subscribe_is_answered_on_the_stream_it_arrived_on() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(subscribe(1))).await.expect("write SUBSCRIBE");
        let mut pr = framed(pr);

        let (msg, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        assert!(matches!(msg, ControlMessage::Subscribe(_)), "{msg:?}");
        assert_eq!(stream.origin(), RequestOrigin::Peer);
        assert_eq!(stream.kind(), RequestKind::Subscribe);
        assert_eq!(stream.request_id().into_inner(), 1);
        assert_eq!(lb.conn.endpoint().peer_request_count(), 1);

        // Section 6.4.2.2: "an endpoint sending a response to a request MUST send
        // the corresponding response message ... before sending a FIN."
        let err = stream.finish().await.err();
        assert!(
            matches!(err, Some(ConnectionError::FinBeforeResponse(1))),
            "a FIN before the response must be refused: {err:?}",
        );

        lb.conn.respond_subscribe_ok(&mut stream, subscribe_ok()).await.expect("respond");
        assert!(stream.responded());
        // On the request's own stream, not the control stream: this reader is
        // the peer's half of the bidirectional stream it opened.
        assert!(matches!(next_control(&mut pr).await, ControlMessage::SubscribeOk(_)));

        // The rest of the same sentence: "the publisher of an Established
        // subscription MUST send PUBLISH_DONE, before sending a FIN."
        assert!(stream.owes_publish_done());
        let err = stream.finish().await.err();
        assert!(
            matches!(err, Some(ConnectionError::FinBeforePublishDone(1))),
            "a FIN before PUBLISH_DONE must be refused: {err:?}",
        );

        lb.conn
            .publish_done_on(&mut stream, v(0), v(0), Vec::new())
            .await
            .expect("publish_done_on");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::PublishDone(_)));

        // And now the FIN, which is a clean end rather than a reset.
        let end = pr.read_control(false).await.expect_err("the stream should have ended");
        assert!(
            matches!(end, ConnectionError::UnexpectedEnd),
            "the completed request stream should end with a FIN, not {end}",
        );
    }

    /// A refused update leaves the stream open for the termination it owes.
    ///
    /// Section 9.5.1: "When a REQUEST_UPDATE is unsuccessful, the publisher
    /// MUST also terminate the subscription by sending a PUBLISH_DONE with
    /// error code UPDATE_FAILED." That message has to go on the request's own
    /// stream, so a REQUEST_ERROR answering an update cannot finish the send
    /// half the way one answering the request does.
    #[tokio::test]
    async fn a_refused_update_leaves_the_stream_open_for_its_termination() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(subscribe(1))).await.expect("write SUBSCRIBE");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        lb.conn.respond_subscribe_ok(&mut stream, subscribe_ok()).await.expect("respond");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::SubscribeOk(_)));
        assert!(stream.owes_publish_done());

        // The peer updates the subscription and this endpoint refuses it.
        // The read is what puts the update on the endpoint's books; refusing
        // one it has not seen is a different error entirely.
        ps.write_all(&encode(request_update(1))).await.expect("write REQUEST_UPDATE");
        let update = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut stream))
            .await
            .expect("the update never arrived")
            .expect("read the update");
        assert!(matches!(update, ControlMessage::RequestUpdate(_)), "{update:?}");
        lb.conn.respond_error(&mut stream, request_error()).await.expect("refuse the update");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestError(_)));

        // The subscription is still live and still owes its ending, so the
        // stream is neither finished nor relieved of the obligation.
        assert!(
            stream.owes_publish_done(),
            "a refused update retires nothing: the subscription still owes a PUBLISH_DONE",
        );

        lb.conn
            .publish_done_on(&mut stream, v(0x8), v(0), Vec::new())
            .await
            .expect("the termination the refusal owes");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::PublishDone(_)));
    }

    /// A refused namespace update closes the stream rather than owing a message.
    ///
    /// Section 9.5.1 sorts a refused REQUEST_UPDATE by what was being
    /// updated, and a namespace subscription falls in the clause that ends
    /// with the transport rather than with a message: "When a REQUEST_UPDATE
    /// fails for a SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS or PUBLISH_NAMESPACE,
    /// the responder MUST close the bidi stream (see Section 6.4.2.2)." There is
    /// no subscription to terminate here and no PUBLISH_DONE that could go
    /// out, so a send half held open for one would stay open for good.
    #[tokio::test]
    async fn a_refused_namespace_update_closes_the_stream() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(ControlMessage::SubscribeNamespace(SubscribeNamespace {
            request_id: v(1),
            namespace_prefix: ns(),
            parameters: vec![],
        })))
        .await
        .expect("write SUBSCRIBE_NAMESPACE");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        lb.conn
            .respond_ok(&mut stream, RequestOk { parameters: vec![], track_properties: vec![] })
            .await
            .expect("respond_ok");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestOk(_)));

        // The peer asks to move the prefix and this endpoint refuses.
        ps.write_all(&encode(request_update(1))).await.expect("write REQUEST_UPDATE");
        let update = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut stream))
            .await
            .expect("read hung")
            .expect("read REQUEST_UPDATE");
        assert!(matches!(update, ControlMessage::RequestUpdate(_)));
        lb.conn.respond_error(&mut stream, request_error()).await.expect("refuse the update");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestError(_)));

        // The refusal ends this request stream, so the peer's next read finds
        // the send half closed rather than waiting on a message a namespace
        // subscription has no way to send.
        let ended = tokio::time::timeout(PATIENCE, pr.read_control(false))
            .await
            .expect("the peer is still reading: the refusal never closed the send half")
            .expect_err("a refused namespace update left the stream open");
        assert!(
            matches!(ended, ConnectionError::UnexpectedEnd),
            "the peer should have seen the FIN, got {ended:?}",
        );
    }

    /// Draft-23 Section 6.4.2.3: "When an endpoint rejects a request without
    /// performing any application processing, it SHOULD send a REQUEST_ERROR
    /// and FIN the stream." The FIN is what tells the peer the request is over;
    /// without it the handle's [`Drop`] would reset the stream instead.
    #[tokio::test]
    async fn a_rejected_request_is_finished_rather_than_reset() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(subscribe(1))).await.expect("write SUBSCRIBE");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        lb.conn.respond_error(&mut stream, request_error()).await.expect("respond_error");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestError(_)));

        // Dropping the handle after the FIN must not turn a rejection into a
        // reset, so the end of stream is read after the handle is gone.
        drop(stream);
        let end = pr.read_control(false).await.expect_err("the stream should have ended");
        assert!(
            matches!(end, ConnectionError::UnexpectedEnd),
            "a rejected request stream should be finished, not {end}",
        );
    }

    /// Draft-23 Section 6.3: a bidirectional stream that begins with a message
    /// type that opens no request "MUST close the Session with UNEXPECTED_MESSAGE" — on the wire,
    /// not only in the local state machine. The code is not the INVALID_REQUEST_ID
    /// of draft-23 Section 6.4.2.1.
    #[tokio::test]
    async fn a_stream_that_opens_no_request_closes_the_session() {
        let mut lb = loopback().await;

        let (mut ps, _pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(ControlMessage::GoAway(GoAway {
            new_session_uri: Vec::new(),
            timeout: v(0),
        })))
        .await
        .expect("write GOAWAY");

        // `RequestStream` is not `Debug` — it owns two live stream halves — so
        // the accepted case is named rather than unwrapped.
        let err = match tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
        {
            Err(e) => e,
            Ok((msg, _)) => panic!("a GOAWAY opens no request stream, but {msg:?} was accepted"),
        };
        assert!(
            matches!(err, ConnectionError::NonRequestOnRequestStream(MessageType::GoAway)),
            "{err}",
        );

        assert_closed_with(&lb.peer, UNEXPECTED_MESSAGE, "GoAway").await;
    }

    /// Draft-23 Section 6.4.2.1: "If an endpoint receives a Request ID where the
    /// least significant bit is incorrect for the sender ... it MUST close the
    /// session with INVALID_REQUEST_ID." A different rule from the one above,
    /// and a different code, so a peer can tell them apart.
    #[tokio::test]
    async fn a_peer_id_with_the_wrong_parity_closes_the_session() {
        let mut lb = loopback().await;

        let (mut ps, _pr) = lb.peer.open_bi().await.expect("open_bi");
        // 2 is even, so it is an id this client allocates for itself.
        ps.write_all(&encode(subscribe(2))).await.expect("write SUBSCRIBE");

        let err = match tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
        {
            Err(e) => e,
            Ok((msg, _)) => {
                panic!("an even Request ID from a server peer must be refused, not {msg:?}")
            }
        };
        assert!(matches!(err, ConnectionError::Endpoint(EndpointError::RequestId(_))), "{err}");

        assert_closed_with(&lb.peer, INVALID_REQUEST_ID, "parity").await;
    }

    /// Dropping a request stream is one act with two meanings, and the peer
    /// can only tell them apart if the codes differ: a request this endpoint
    /// made was cancelled, a request made of it was never served.
    #[tokio::test]
    async fn an_unserved_request_and_a_cancelled_one_reset_with_different_codes() {
        let mut lb = loopback().await;

        // Inbound: the peer asked, and the handle was dropped unanswered.
        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(subscribe(1))).await.expect("write SUBSCRIBE");
        let mut pr = framed(pr);
        let (_, stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        drop(stream);
        let err = tokio::time::timeout(PATIENCE, pr.read_control(false))
            .await
            .expect("the inbound stream was neither reset nor answered")
            .expect_err("the dropped handle should have reset the stream");
        assert!(
            matches!(
                err,
                ConnectionError::Transport(TransportError::StreamReset(c))
                    if c == REQUEST_UNANSWERED
            ),
            "an unserved inbound request should reset with INTERNAL_ERROR (0x0): {err}",
        );

        // Outbound: this endpoint asked, and walked away.
        let outbound = lb.conn.subscribe(ns(), b"audio".to_vec(), vec![]).await.expect("subscribe");
        let (_, their_recv) = tokio::time::timeout(PATIENCE, lb.peer.accept_bi())
            .await
            .expect("the client opened no request stream")
            .expect("accept_bi");
        let mut their_recv = framed(their_recv);
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::Subscribe(_)));
        drop(outbound);
        let err = tokio::time::timeout(PATIENCE, their_recv.read_control(false))
            .await
            .expect("the outbound stream was never reset")
            .expect_err("the dropped handle should have reset the stream");
        assert!(
            matches!(
                err,
                ConnectionError::Transport(TransportError::StreamReset(c))
                    if c == REQUEST_CANCELLED
            ),
            "an abandoned outbound request should reset with CANCELLED (0x1): {err}",
        );
    }

    /// A dropped accept future loses neither the stream nor the bytes of the
    /// request that had already arrived, which is what makes
    /// [`Connection::accept_request_stream`] safe to `select!` against anything
    /// else.
    #[tokio::test]
    async fn a_cancelled_accept_keeps_the_stream_and_what_had_arrived() {
        let mut lb = loopback().await;

        let (mut ps, _pr) = lb.peer.open_bi().await.expect("open_bi");
        let bytes = encode(subscribe(1));
        // One byte: enough to open the stream and to be read as the start of a
        // message, not enough to complete one.
        ps.write_all(&bytes[..1]).await.expect("write the first byte");

        assert!(
            tokio::time::timeout(QUIET, lb.conn.accept_request_stream()).await.is_err(),
            "a partial request must not complete an accept",
        );
        assert_eq!(
            lb.conn.pending_inbound_count(),
            1,
            "the cancelled accept dropped the peer's stream",
        );

        ps.write_all(&bytes[1..]).await.expect("write the rest");
        let (msg, stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("the retried accept hung")
            .expect("accept");
        // The Request ID could not be read at all if the first byte had gone
        // with the cancelled future: the message would not decode.
        assert_eq!(stream.request_id().into_inner(), 1);
        assert!(matches!(msg, ControlMessage::Subscribe(_)), "{msg:?}");
        assert_eq!(lb.conn.pending_inbound_count(), 0);
    }

    /// The `respond_*` helpers answer requests the peer made, and refuse the
    /// ones this endpoint made. The consequence is that nothing is written:
    /// the peer sees the request and then silence, not a response to itself.
    #[tokio::test]
    async fn a_respond_helper_writes_nothing_on_a_stream_this_endpoint_opened() {
        let mut lb = loopback().await;

        let mut outbound =
            lb.conn.subscribe(ns(), b"video".to_vec(), vec![]).await.expect("subscribe");
        let (_, their_recv) = tokio::time::timeout(PATIENCE, lb.peer.accept_bi())
            .await
            .expect("the client opened no request stream")
            .expect("accept_bi");
        let mut their_recv = framed(their_recv);
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::Subscribe(_)));

        let err = lb
            .conn
            .respond_error(&mut outbound, request_error())
            .await
            .expect_err("a request this endpoint made is not ours to answer");
        assert!(matches!(err, ConnectionError::RespondedToOwnRequest(0)), "{err}");

        assert!(
            tokio::time::timeout(QUIET, their_recv.read_control(false)).await.is_err(),
            "the refused response reached the wire anyway",
        );
    }

    /// A refused fetch update resets the stream the objects were going out on.
    ///
    /// Section 9.5.1: "When a REQUEST_UPDATE fails for a FETCH, the publisher
    /// MUST reset the FETCH data stream." The objects are not on the request
    /// stream, so refusing the update on that stream is only half of it; the
    /// data stream has to go too, and the connection can only reset a handle
    /// it still holds.
    #[tokio::test]
    async fn a_refused_fetch_update_resets_the_fetch_data_stream() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(peer_fetch(1))).await.expect("write FETCH");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        assert_eq!(stream.kind(), RequestKind::Fetch);
        lb.conn.respond_ok(&mut stream, fetch_ok()).await.expect("respond_ok");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestOk(_)));

        // The objects go out on a stream of their own, which the request now
        // holds.
        lb.conn
            .open_fetch_stream_on(
                &mut stream,
                &AnyFetchHeader::Draft23(FetchHeader { request_id: v(1) }),
            )
            .await
            .expect("open the fetch data stream");
        let data = tokio::time::timeout(PATIENCE, lb.peer.accept_uni())
            .await
            .expect("no data stream arrived")
            .expect("accept_uni");
        let mut data = framed(data);
        tokio::time::timeout(PATIENCE, data.read_fetch_header())
            .await
            .expect("the header never arrived")
            .expect("the data stream opens with a FETCH_HEADER");

        // The peer updates the fetch and this endpoint refuses it.
        ps.write_all(&encode(request_update(1))).await.expect("write REQUEST_UPDATE");
        tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut stream))
            .await
            .expect("read hung")
            .expect("read REQUEST_UPDATE");
        lb.conn.respond_error(&mut stream, request_error()).await.expect("refuse the update");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestError(_)));

        // The data stream went with it: the peer's next read on it fails
        // rather than waiting for objects that are not coming.
        let err = tokio::time::timeout(PATIENCE, data.read_control(false))
            .await
            .expect("the peer is still waiting on a stream that should have been reset")
            .expect_err("the data stream should have been reset");
        assert!(
            format!("{err}").to_lowercase().contains("reset"),
            "the peer should see a reset, got {err}",
        );
    }

    /// A publisher can open the stream a FETCH's objects go on, and owns it.
    ///
    /// The objects answering a FETCH go on a unidirectional stream of their
    /// own that opens with a FETCH_HEADER.
    #[tokio::test]
    async fn a_fetch_data_stream_can_be_opened_through_the_connection() {
        let lb = loopback().await;

        let mut send = lb
            .conn
            .open_fetch_stream(&AnyFetchHeader::Draft23(FetchHeader { request_id: v(4) }))
            .await
            .expect("open a fetch data stream");

        let recv = tokio::time::timeout(PATIENCE, lb.peer.accept_uni())
            .await
            .expect("the client opened no data stream")
            .expect("accept_uni");
        let mut recv = framed(recv);
        let header = tokio::time::timeout(PATIENCE, recv.read_fetch_header())
            .await
            .expect("the header never arrived")
            .expect("the stream opens with a FETCH_HEADER");
        assert!(
            matches!(&header, AnyFetchHeader::Draft23(h) if h.request_id.into_inner() == 4),
            "the header names the request it answers: {header:?}",
        );

        // The caller holds the stream, which is what lets a request that fails
        // later reset the objects it was serving.
        send.reset(0).expect("reset the fetch data stream");
    }

    /// Draft-23 Table 5 gives NAMESPACE the Stream value "Request", and
    /// Section 9.15 puts it "on the response stream of a SUBSCRIBE_NAMESPACE
    /// request". It is checked by reading the peer's own half of the request
    /// stream, which a control-stream write could not reach.
    #[tokio::test]
    async fn a_namespace_is_announced_on_the_request_stream_that_asked_for_it() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(ControlMessage::SubscribeNamespace(SubscribeNamespace {
            request_id: v(1),
            namespace_prefix: ns(),
            parameters: vec![],
        })))
        .await
        .expect("write SUBSCRIBE_NAMESPACE");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        assert_eq!(stream.kind(), RequestKind::SubscribeNamespace);

        lb.conn
            .respond_ok(&mut stream, RequestOk { parameters: vec![], track_properties: vec![] })
            .await
            .expect("respond_ok");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestOk(_)));

        lb.conn
            .namespace_on(&mut stream, Namespace { namespace_suffix: ns(), parameters: vec![] })
            .await
            .expect("namespace_on");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::Namespace(_)));

        lb.conn
            .namespace_done_on(&mut stream, NamespaceDone { namespace_suffix: ns() })
            .await
            .expect("namespace_done_on");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::NamespaceDone(_)));

        // Nothing further is owed on this direction, so the responder may
        // finish it — Section 6.4.2.2's guard covers the response and a
        // publisher's PUBLISH_DONE, not announcements that may never come.
        stream.finish().await.expect("finish an answered namespace subscription");
    }

    /// An update on a PUBLISH this endpoint sent is answered here.
    ///
    /// Section 9.5 names the one case where a requester answers rather than
    /// asks: "A subscriber can also send REQUEST_UPDATE to modify parameters
    /// of a subscription established with PUBLISH." The receiver of that
    /// update "MUST respond with exactly one REQUEST_UPDATE_OK or
    /// REQUEST_UPDATE_ERROR message indicating if the update was successful" —
    /// a REQUEST_OK or a REQUEST_ERROR, under Section 1.5's names — and on a PUBLISH this
    /// endpoint sent, the receiver is this endpoint.
    #[tokio::test]
    async fn an_update_on_a_publish_we_sent_is_answered_here() {
        let mut lb = loopback().await;

        let mut outbound =
            lb.conn.publish(ns(), b"video".to_vec(), v(7), vec![], vec![]).await.expect("publish");
        let (mut their_send, their_recv) = tokio::time::timeout(PATIENCE, lb.peer.accept_bi())
            .await
            .expect("the client opened no request stream")
            .expect("accept_bi");
        let mut their_recv = framed(their_recv);
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::Publish(_)));

        // The subscriber accepts the publication, then updates it.
        their_send
            .write_all(&encode(ControlMessage::RequestOk(RequestOk {
                parameters: vec![],
                track_properties: vec![],
            })))
            .await
            .expect("write REQUEST_OK");
        let msg = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut outbound))
            .await
            .expect("no REQUEST_OK arrived")
            .expect("read REQUEST_OK");
        assert!(matches!(msg, ControlMessage::RequestOk(_)), "{msg:?}");

        their_send.write_all(&encode(request_update(0))).await.expect("write REQUEST_UPDATE");
        let msg = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut outbound))
            .await
            .expect("no REQUEST_UPDATE arrived")
            .expect("read REQUEST_UPDATE");
        assert!(matches!(msg, ControlMessage::RequestUpdate(_)), "{msg:?}");

        lb.conn
            .respond_ok(&mut outbound, RequestOk { parameters: vec![], track_properties: vec![] })
            .await
            .expect("the subscriber's update is this endpoint's to answer");
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::RequestOk(_)));

        // The publication is untouched by the update and still owes its ending.
        assert!(outbound.owes_publish_done());
        lb.conn
            .publish_done(&mut outbound, v(0), v(0), Vec::new())
            .await
            .expect("an accepted update leaves the ending free");
    }

    /// Refusing that update owes the same ending as refusing any other.
    ///
    /// Section 9.5.1: "When a REQUEST_UPDATE is unsuccessful, the publisher
    /// MUST also terminate the subscription by sending a PUBLISH_DONE with
    /// error code UPDATE_FAILED." The publisher of a subscription established
    /// with PUBLISH is the endpoint that sent it, and the ending goes on the
    /// stream that endpoint opened rather than on one the peer opened.
    #[tokio::test]
    async fn a_refused_update_on_a_publish_we_sent_owes_its_ending() {
        let mut lb = loopback().await;

        let mut outbound =
            lb.conn.publish(ns(), b"video".to_vec(), v(7), vec![], vec![]).await.expect("publish");
        let (mut their_send, their_recv) = tokio::time::timeout(PATIENCE, lb.peer.accept_bi())
            .await
            .expect("the client opened no request stream")
            .expect("accept_bi");
        let mut their_recv = framed(their_recv);
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::Publish(_)));

        their_send
            .write_all(&encode(ControlMessage::RequestOk(RequestOk {
                parameters: vec![],
                track_properties: vec![],
            })))
            .await
            .expect("write REQUEST_OK");
        tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut outbound))
            .await
            .expect("no REQUEST_OK arrived")
            .expect("read REQUEST_OK");

        their_send.write_all(&encode(request_update(0))).await.expect("write REQUEST_UPDATE");
        tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut outbound))
            .await
            .expect("no REQUEST_UPDATE arrived")
            .expect("read REQUEST_UPDATE");

        lb.conn
            .respond_error(&mut outbound, request_error())
            .await
            .expect("refuse the subscriber's update");
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::RequestError(_)));
        assert!(
            outbound.owes_publish_done(),
            "a refused update retires nothing: the subscription still owes a PUBLISH_DONE",
        );

        // The ending is owed under one status, and asking for another leaves
        // the publication exactly where it was rather than half ended.
        let err = lb
            .conn
            .publish_done(&mut outbound, v(0), v(0), Vec::new())
            .await
            .expect_err("a refused update fixes the status of the ending");
        assert!(
            matches!(
                err,
                ConnectionError::Endpoint(EndpointError::WrongUpdateFailureStatus {
                    request: 0,
                    required: 0x8,
                })
            ),
            "{err}",
        );
        lb.conn
            .publish_done(&mut outbound, v(0x8), v(0), Vec::new())
            .await
            .expect("the termination the refusal owes");
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::PublishDone(_)));
    }

    /// Draft-23 Section 6.4.2.2 excludes the sender of a PUBLISH from the
    /// requesters that "MAY FIN immediately after sending a message", because
    /// an accepted PUBLISH makes it the publisher of an Established
    /// subscription. A **rejected** one establishes nothing, so the obligation
    /// has to end with the REQUEST_ERROR — otherwise the only way to close a
    /// refused publication would be to reset the stream, telling the peer the
    /// request was abandoned rather than answered.
    #[tokio::test]
    async fn a_refused_publish_can_still_be_finished_cleanly() {
        let mut lb = loopback().await;

        let mut outbound =
            lb.conn.publish(ns(), b"video".to_vec(), v(7), vec![], vec![]).await.expect("publish");
        assert!(outbound.owes_publish_done(), "a PUBLISH sender owes a PUBLISH_DONE");
        let err = outbound.finish().await.err();
        assert!(
            matches!(err, Some(ConnectionError::FinBeforePublishDone(0))),
            "a PUBLISH may not be finished while its subscription could still be established: \
             {err:?}",
        );

        let (mut their_send, their_recv) = tokio::time::timeout(PATIENCE, lb.peer.accept_bi())
            .await
            .expect("the client opened no request stream")
            .expect("accept_bi");
        let mut their_recv = framed(their_recv);
        assert!(matches!(next_control(&mut their_recv).await, ControlMessage::Publish(_)));

        their_send
            .write_all(&encode(ControlMessage::RequestError(request_error())))
            .await
            .expect("write REQUEST_ERROR");
        let msg = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut outbound))
            .await
            .expect("no REQUEST_ERROR arrived")
            .expect("a refused PUBLISH is an ordinary response");
        assert!(matches!(msg, ControlMessage::RequestError(_)), "{msg:?}");
        assert!(!outbound.owes_publish_done());

        outbound.finish().await.expect("a refused publication owes nothing further");
        drop(outbound);
        let end = their_recv.read_control(false).await.expect_err("the stream should have ended");
        assert!(
            matches!(end, ConnectionError::UnexpectedEnd),
            "a refused publication should end with a FIN, not {end}",
        );
    }

    /// A peer that PUBLISHes to us is the publisher, so PUBLISH_DONE arrives
    /// on the stream it opened rather than being sent on it. Routing a
    /// peer-opened stream's reads through the response dispatcher would look
    /// up a request this endpoint never made.
    #[tokio::test]
    async fn a_peers_publish_is_ended_by_the_peer_on_its_own_stream() {
        let mut lb = loopback().await;

        let (mut ps, pr) = lb.peer.open_bi().await.expect("open_bi");
        ps.write_all(&encode(ControlMessage::Publish(Publish {
            request_id: v(1),
            track_namespace: ns(),
            track_name: b"video".to_vec(),
            track_alias: v(7),
            parameters: vec![],
            track_properties: vec![],
        })))
        .await
        .expect("write PUBLISH");
        let mut pr = framed(pr);

        let (_, mut stream) = tokio::time::timeout(PATIENCE, lb.conn.accept_request_stream())
            .await
            .expect("accept hung")
            .expect("accept");
        assert_eq!(stream.kind(), RequestKind::Publish);

        lb.conn
            .respond_ok(&mut stream, RequestOk { parameters: vec![], track_properties: vec![] })
            .await
            .expect("respond_ok");
        assert!(matches!(next_control(&mut pr).await, ControlMessage::RequestOk(_)));

        ps.write_all(&encode(ControlMessage::PublishDone(PublishDone {
            status_code: v(0),
            stream_count: v(0),
            reason_phrase: Vec::new(),
        })))
        .await
        .expect("write PUBLISH_DONE");

        let msg = tokio::time::timeout(PATIENCE, lb.conn.recv_on_request_stream(&mut stream))
            .await
            .expect("no PUBLISH_DONE arrived")
            .expect("the publishing peer may end its own publication");
        assert!(matches!(msg, ControlMessage::PublishDone(_)), "{msg:?}");
    }
}