micro_h2/

conn.rs

//! A client connection: one HPACK context, a few streams, and flow control.
//!
//! # Flow control is the part that bites
//!
//! An HTTP/2 receiver advertises how much it is willing to accept, and a sender
//! that has used up that allowance simply stops. Both windows start at 65535
//! bytes — one for the connection, one for each stream — and neither grows
//! unless the receiver says so with `WINDOW_UPDATE`.
//!
//! For a client that fetches small responses this never comes up, which is
//! exactly why it is dangerous: the netmap long-poll streams megabytes, and a
//! client that never sends `WINDOW_UPDATE` receives precisely 65535 bytes and
//! then hangs forever. It looks like the server stopped talking. It did, because
//! we told it to.
//!
//! So this connection raises the connection window immediately after the
//! preface, advertises a large per-stream window in its `SETTINGS`, and returns
//! a `WINDOW_UPDATE` for every byte of `DATA` it consumes.
//!
//! # Sans-io
//!
//! The caller reads [`frame::HEADER_LEN`] bytes, learns the payload length,
//! reads that, and hands the whole frame to [`Connection::recv`]. Anything that
//! must be sent in reply is written to the caller's buffer.

use crate::frame::{self, FrameHeader, FrameType, flags, settings};
use crate::hpack::{self, Decoder};
use crate::{Error, headers, hpack::encode};

/// How many requests may be in flight at once. Registration and the map
/// long-poll, with one spare.
pub const MAX_STREAMS: usize = 4;

/// The largest header block this client will reassemble across `CONTINUATION`
/// frames.
pub const MAX_HEADER_BLOCK: usize = 2048;

/// What we advertise as our per-stream receive window, and what we top the
/// connection window up to.
///
/// Large enough that a streaming response is never throttled by the round trip
/// of a `WINDOW_UPDATE`, and it costs nothing: the window bounds what the peer
/// may have in flight, not what we must buffer, because every `DATA` frame is
/// handed to the caller as it arrives.
pub const RECEIVE_WINDOW: u32 = 1 << 20;

/// The default both windows start at, before anything is negotiated.
const DEFAULT_WINDOW: u32 = 65_535;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Stream {
    id: u32,
    /// Bytes received since the last `WINDOW_UPDATE` for this stream.
    consumed: u32,
    open: bool,
    send_window: i64,
    head_request: bool,
    connect_request: bool,
    response_started: bool,
    no_body: bool,
    content_length: Option<u64>,
    received: u64,
}

/// What arrived.
#[derive(Debug, PartialEq, Eq)]
pub enum Event<'a> {
    /// Nothing the caller needs to act on: a settings exchange, a ping, an
    /// unknown frame type, or a late frame on a closed stream. Anything that
    /// needed a reply is already in `out`.
    Nothing,
    /// A complete header block. The fields were passed to the callback.
    Headers { stream: u32, end_stream: bool },
    /// Response body bytes. Borrowed from the caller's own frame buffer, so
    /// nothing is copied.
    Data {
        stream: u32,
        data: &'a [u8],
        end_stream: bool,
    },
    /// The peer reset one stream. The connection is still usable.
    Reset { stream: u32, code: u32 },
    /// The peer is shutting the connection down.
    GoAway { code: u32 },
}

pub struct Connection {
    decoder: Decoder,
    streams: heapless::Vec<Stream, MAX_STREAMS>,
    next_stream: u32,
    /// Reassembly buffer for a header block split across `CONTINUATION` frames.
    header_block: heapless::Vec<u8, MAX_HEADER_BLOCK>,
    /// Which stream the block being reassembled belongs to.
    header_stream: u32,
    header_end_stream: bool,
    /// Bytes received on the connection since the last `WINDOW_UPDATE`.
    consumed: u32,
    /// What the peer said it will accept in one frame.
    peer_max_frame: usize,
    send_window: u32,
    peer_initial_window: u32,
    peer_max_streams: u32,
    goaway: bool,
}

impl Default for Connection {
    fn default() -> Self {
        Self::new()
    }
}

impl Connection {
    pub fn new() -> Self {
        Self {
            decoder: Decoder::new(hpack::DEFAULT_TABLE_SIZE),
            streams: heapless::Vec::new(),
            // Client-initiated streams are odd-numbered.
            next_stream: 1,
            header_block: heapless::Vec::new(),
            header_stream: 0,
            header_end_stream: false,
            consumed: 0,
            peer_max_frame: frame::DEFAULT_MAX_FRAME,
            send_window: DEFAULT_WINDOW,
            peer_initial_window: DEFAULT_WINDOW,
            peer_max_streams: u32::MAX,
            goaway: false,
        }
    }

    /// Write the client preface, our settings, and the connection-level window
    /// update that stops a large response stalling at 65535 bytes.
    pub fn start(&mut self, out: &mut [u8]) -> Result<usize, Error> {
        let preface = frame::CLIENT_PREFACE;
        out.get_mut(..preface.len())
            .ok_or(Error::BufferTooSmall)?
            .copy_from_slice(preface);
        let mut len = preface.len();

        // Push is refused outright rather than handled: a client that never
        // accepts a promise needs no state for one.
        let mut payload = [0u8; 24];
        write_setting(
            &mut payload[0..6],
            settings::HEADER_TABLE_SIZE,
            hpack::DEFAULT_TABLE_SIZE as u32,
        );
        write_setting(&mut payload[6..12], settings::ENABLE_PUSH, 0);
        write_setting(
            &mut payload[12..18],
            settings::INITIAL_WINDOW_SIZE,
            RECEIVE_WINDOW,
        );
        write_setting(
            &mut payload[18..24],
            settings::MAX_CONCURRENT_STREAMS,
            MAX_STREAMS as u32,
        );
        len += frame::write_frame(FrameType::Settings, 0, 0, &payload, &mut out[len..])?;

        // SETTINGS_INITIAL_WINDOW_SIZE applies to streams only; the connection
        // window can be raised only by an explicit update.
        let increment = (RECEIVE_WINDOW - DEFAULT_WINDOW).to_be_bytes();
        len += frame::write_frame(FrameType::WindowUpdate, 0, 0, &increment, &mut out[len..])?;
        Ok(len)
    }

    /// Open a stream and send a request, headers and body together.
    ///
    /// Returns the stream identifier and how many bytes were written.
    /// The authority argument is authoritative; redundant `host` fields are
    /// omitted. If it is empty, a single `host` field supplies `:authority`.
    #[allow(clippy::too_many_arguments)]
    pub fn request(
        &mut self,
        method: &str,
        path: &str,
        authority: &str,
        scheme: &str,
        extra: &[(&str, &str)],
        body: &[u8],
        out: &mut [u8],
    ) -> Result<(u32, usize), Error> {
        if self.goaway {
            return Err(Error::GoAway);
        }
        let id = self.next_stream;
        if self.streams.len() as u32 >= self.peer_max_streams {
            return Err(Error::TooManyStreams);
        }
        if self.streams.is_full() || id > 0x7fff_ffff {
            return Err(Error::TooManyStreams);
        }

        if method.is_empty()
            || !method.bytes().all(headers::token)
            || (method != "CONNECT"
                && (!headers::path_valid(method, path) || !headers::scheme_valid(scheme)))
        {
            return Err(Error::Protocol);
        }
        let mut content_length = None;
        let mut host = None;
        for &(name, value) in extra {
            headers::field(name, value, true)?;
            if name == "host" && authority.is_empty() && host.replace(value).is_some() {
                return Err(Error::Protocol);
            }
            if name == "content-length" {
                headers::content_length(value, &mut content_length)?;
            }
        }
        let authority = if authority.is_empty() {
            host.unwrap_or("")
        } else {
            authority
        };
        if !headers::authority_valid(authority, scheme, method == "CONNECT") {
            return Err(Error::Protocol);
        }
        if content_length.is_some_and(|length| length != body.len() as u64) {
            return Err(Error::Protocol);
        }

        // Pseudo-headers must come first and in this order; a server is entitled
        // to reject a block that interleaves them with ordinary fields.
        let mut block = [0u8; MAX_HEADER_BLOCK];
        let mut block_len = 0;
        block_len = encode::encode_header(":method", method, &mut block, block_len)?;
        if method != "CONNECT" {
            block_len = encode::encode_header(":path", path, &mut block, block_len)?;
            block_len = encode::encode_header(":scheme", scheme, &mut block, block_len)?;
        }
        if !authority.is_empty() {
            block_len = encode::encode_header(":authority", authority, &mut block, block_len)?;
        }
        let mut length_written = false;
        for (name, value) in extra {
            // Direct HTTP/2 clients convey routing through :authority. Omit
            // Host so callers cannot emit conflicting routing identities.
            if *name == "host" {
                continue;
            }
            if *name == "content-length" {
                if length_written {
                    continue;
                }
                // Emit the already-validated length once in canonical form.
                // Receivers need not normalize duplicate/list forms themselves.
                use core::fmt::Write;
                let mut canonical = heapless::String::<20>::new();
                write!(&mut canonical, "{}", body.len()).map_err(|_| Error::BufferTooSmall)?;
                block_len = encode::encode_header(name, &canonical, &mut block, block_len)?;
                length_written = true;
            } else {
                let value = if *name == "te" { "trailers" } else { value };
                block_len = encode::encode_header(name, value, &mut block, block_len)?;
            }
        }

        // No CONTINUATION is emitted: a header block that does not fit one frame
        // would need one, and these two request shapes never come close.
        if block_len > self.peer_max_frame {
            return Err(Error::FrameTooLarge);
        }

        if body.len() > self.peer_max_frame {
            return Err(Error::FrameTooLarge);
        }
        if body.len() > self.send_window as usize || body.len() > self.peer_initial_window as usize
        {
            return Err(Error::FlowControl);
        }
        let required = frame::HEADER_LEN
            + block_len
            + if body.is_empty() {
                0
            } else {
                frame::HEADER_LEN + body.len()
            };
        if out.len() < required {
            return Err(Error::BufferTooSmall);
        }

        let end_stream = if body.is_empty() {
            flags::END_STREAM
        } else {
            0
        };
        let mut len = frame::write_frame(
            FrameType::Headers,
            flags::END_HEADERS | end_stream,
            id,
            &block[..block_len],
            out,
        )?;

        if !body.is_empty() {
            len += frame::write_frame(
                FrameType::Data,
                flags::END_STREAM,
                id,
                body,
                &mut out[len..],
            )?;
        }
        self.streams
            .push(Stream {
                id,
                consumed: 0,
                open: true,
                send_window: self.peer_initial_window as i64 - body.len() as i64,
                head_request: method == "HEAD",
                connect_request: method == "CONNECT",
                response_started: false,
                no_body: false,
                content_length: None,
                received: 0,
            })
            .map_err(|_| Error::TooManyStreams)?;
        self.next_stream += 2;
        self.send_window -= body.len() as u32;
        Ok((id, len))
    }

    /// Consume one whole frame.
    ///
    /// `frame` is the nine-byte header followed by exactly its payload. Replies
    /// this connection owes — settings and ping acknowledgements, window updates
    /// — are written to `out`, and the caller must send them.
    pub fn recv<'a>(
        &mut self,
        bytes: &'a [u8],
        mut on_header: impl FnMut(&str, &str),
        out: &mut [u8],
    ) -> Result<(Event<'a>, usize), Error> {
        let header = FrameHeader::parse(bytes)?;
        let payload = bytes
            .get(frame::HEADER_LEN..frame::HEADER_LEN + header.length)
            .ok_or(Error::Incomplete)?;
        if header.length > frame::DEFAULT_MAX_FRAME {
            return Err(Error::FrameTooLarge);
        }
        if self.header_stream != 0 {
            if header.kind != FrameType::Continuation || header.stream != self.header_stream {
                return Err(Error::Protocol);
            }
        } else if header.kind == FrameType::Continuation {
            return Err(Error::Protocol);
        }
        match header.kind {
            FrameType::Settings
                if header.stream != 0 || (header.has(flags::ACK) && !payload.is_empty()) =>
            {
                return Err(Error::Protocol);
            }
            FrameType::Ping if header.stream != 0 || payload.len() != 8 => {
                return Err(Error::Protocol);
            }
            FrameType::GoAway if header.stream != 0 || payload.len() < 8 => {
                return Err(Error::Protocol);
            }
            FrameType::Headers | FrameType::Data | FrameType::Continuation
                if header.stream == 0 =>
            {
                return Err(Error::Protocol);
            }
            FrameType::RstStream if header.stream == 0 || payload.len() != 4 => {
                return Err(Error::Protocol);
            }
            FrameType::Priority if header.stream == 0 || payload.len() != 5 => {
                return Err(Error::Protocol);
            }
            FrameType::WindowUpdate
                if payload.len() != 4 || read_u32(payload).unwrap() & 0x7fff_ffff == 0 =>
            {
                return Err(Error::Protocol);
            }
            _ => {}
        }
        if matches!(
            header.kind,
            FrameType::Headers | FrameType::Data | FrameType::RstStream | FrameType::WindowUpdate
        ) && header.stream != 0
        {
            self.known_stream(header.stream)?;
        }
        let mut written = 0;

        match header.kind {
            FrameType::Settings => {
                if !header.has(flags::ACK) {
                    self.apply_settings(payload)?;
                    // An unacknowledged SETTINGS blocks the peer indefinitely.
                    written = frame::write_frame(FrameType::Settings, flags::ACK, 0, &[], out)?;
                }
                Ok((Event::Nothing, written))
            }

            FrameType::Ping => {
                if !header.has(flags::ACK) {
                    // The payload must be echoed exactly.
                    written = frame::write_frame(FrameType::Ping, flags::ACK, 0, payload, out)?;
                }
                Ok((Event::Nothing, written))
            }

            FrameType::Headers | FrameType::Continuation => {
                let block = if header.kind == FrameType::Headers {
                    self.header_block.clear();
                    self.header_stream = header.stream;
                    self.header_end_stream = header.has(flags::END_STREAM);
                    let body = frame::strip_padding(payload, header.flags)?;
                    // A priority block sits between the padding and the header
                    // block, and feeding it to HPACK fails on a valid frame.
                    if header.has(flags::PRIORITY) {
                        body.get(5..).ok_or(Error::Protocol)?
                    } else {
                        body
                    }
                } else {
                    payload
                };

                self.header_block
                    .extend_from_slice(block)
                    .map_err(|_| Error::BufferTooSmall)?;

                if !header.has(flags::END_HEADERS) {
                    // More CONTINUATION frames to come. Decoding a partial block
                    // would corrupt the HPACK state for the whole connection.
                    return Ok((Event::Nothing, 0));
                }

                let stream = self.header_stream;
                let end_stream = self.header_end_stream;
                let active = self.streams.iter().position(|entry| entry.id == stream);
                if let Some(index) = active {
                    if self.streams[index].response_started {
                        // Trailers remain outside this client's supported surface.
                        return Err(Error::Protocol);
                    }
                    // Validate the whole block before any fields escape to the caller.
                    // The fixed-size clone keeps the real HPACK table at its starting
                    // state for the delivery pass; each insertion commits only once.
                    let mut preflight = self.decoder.clone();
                    let mut fields = headers::ResponseHeaders::default();
                    let mut valid = Ok(());
                    preflight.decode(&self.header_block, |name, value| {
                        if valid.is_ok() {
                            valid = fields.field(name, value);
                        }
                    })?;
                    valid?;
                    let status = fields.status.ok_or(Error::Protocol)?;
                    if status < 200 {
                        if end_stream || fields.content_length.is_some() {
                            return Err(Error::Protocol);
                        }
                    } else {
                        let entry = &mut self.streams[index];
                        let tunnel = entry.connect_request && (200..300).contains(&status);
                        // RFC 9110 section 9.3.6 requires clients to ignore a
                        // successful CONNECT's Content-Length. Ordinary 205
                        // has zero-length content, unlike the HEAD/304 metadata
                        // exceptions in sections 6.4.1 and 8.6.
                        if !tunnel
                            && ((status == 204 && fields.content_length.is_some())
                                || (status == 205
                                    && fields.content_length.is_some_and(|length| length != 0)))
                        {
                            return Err(Error::Protocol);
                        }
                        let no_body =
                            !tunnel && (entry.head_request || matches!(status, 204 | 205 | 304));
                        let expected = if no_body || tunnel {
                            None
                        } else {
                            fields.content_length
                        };
                        if end_stream && expected.is_some_and(|length| length != 0) {
                            return Err(Error::Protocol);
                        }
                        entry.response_started = true;
                        entry.no_body = no_body;
                        entry.content_length = expected;
                    }
                    self.decoder.decode(&self.header_block, &mut on_header)?;
                } else {
                    // Late fields on closed streams still affect connection-wide
                    // HPACK state, but must not become application headers.
                    self.decoder.decode(&self.header_block, |_, _| {})?;
                }
                self.header_stream = 0;
                self.header_block.clear();
                if active.is_none() {
                    return Ok((Event::Nothing, 0));
                }
                if end_stream {
                    self.close(stream);
                }
                Ok((Event::Headers { stream, end_stream }, 0))
            }

            FrameType::Data => {
                let data = frame::strip_padding(payload, header.flags)?;
                let end_stream = header.has(flags::END_STREAM);
                let active = self
                    .streams
                    .iter()
                    .position(|entry| entry.id == header.stream);
                let received = active
                    .map(|index| self.data_length(index, data.len(), end_stream))
                    .transpose()?;
                // The window is consumed by the whole payload, padding included,
                // not just the bytes we hand back.
                written = self.credit(header.stream, header.length as u32, out)?;
                let Some(index) = active else {
                    return Ok((Event::Nothing, written));
                };
                self.streams[index].received = received.unwrap();
                if end_stream {
                    self.close(header.stream);
                }
                Ok((
                    Event::Data {
                        stream: header.stream,
                        data,
                        end_stream,
                    },
                    written,
                ))
            }

            FrameType::RstStream => {
                let code = read_u32(payload).ok_or(Error::Protocol)?;
                if !self.streams.iter().any(|entry| entry.id == header.stream) {
                    return Ok((Event::Nothing, 0));
                }
                self.close(header.stream);
                Ok((
                    Event::Reset {
                        stream: header.stream,
                        code,
                    },
                    0,
                ))
            }

            FrameType::GoAway => {
                // last-stream-id, then the error code.
                let code = payload
                    .get(4..8)
                    .and_then(read_u32)
                    .ok_or(Error::Protocol)?;
                self.goaway = true;
                Ok((Event::GoAway { code }, 0))
            }

            FrameType::WindowUpdate => {
                let increment = read_u32(payload).ok_or(Error::Protocol)? & 0x7fff_ffff;
                if header.stream == 0 {
                    self.send_window = self
                        .send_window
                        .checked_add(increment)
                        .filter(|value| *value <= 0x7fff_ffff)
                        .ok_or(Error::Protocol)?;
                } else if let Some(stream) = self.streams.iter_mut().find(|s| s.id == header.stream)
                {
                    let window = stream.send_window + increment as i64;
                    if window > 0x7fff_ffff {
                        return Err(Error::Protocol);
                    }
                    stream.send_window = window;
                }
                Ok((Event::Nothing, 0))
            }
            FrameType::Priority => Ok((Event::Nothing, 0)),

            // Push was refused in our SETTINGS, so a promise is a protocol
            // violation rather than something to ignore.
            FrameType::PushPromise => Err(Error::Protocol),

            // RFC 7540 requires unknown types to be discarded.
            FrameType::Unknown(_) => Ok((Event::Nothing, 0)),
        }
    }

    /// Complete a DATA frame whose payload was consumed incrementally by the
    /// transport adapter.
    ///
    /// DATA does not affect HPACK state, so constrained adapters can stream a
    /// large payload without retaining the complete frame. Flow-control credit
    /// and stream closure still happen exactly once, after every payload byte
    /// has been authenticated and consumed. For padded frames, use
    /// [`Self::finish_data_with_length`] to supply the unpadded content length.
    pub fn finish_data(&mut self, header: FrameHeader, out: &mut [u8]) -> Result<usize, Error> {
        if header.has(flags::PADDED) {
            return Err(Error::Protocol);
        }
        self.finish_data_with_length(header, header.length, out)
    }

    /// Complete a streamed DATA frame, supplying the unpadded content length.
    /// The adapter must validate the padding byte and consume the full payload
    /// before calling this method. Credit includes padding; Content-Length does not.
    pub fn finish_data_with_length(
        &mut self,
        header: FrameHeader,
        data_length: usize,
        out: &mut [u8],
    ) -> Result<usize, Error> {
        if header.kind != FrameType::Data || header.stream == 0 || self.header_stream != 0 {
            return Err(Error::Protocol);
        }
        if header.length > frame::DEFAULT_MAX_FRAME {
            return Err(Error::FrameTooLarge);
        }
        let padding = header
            .length
            .checked_sub(data_length)
            .ok_or(Error::Protocol)?;
        if (header.has(flags::PADDED) && !(1..=256).contains(&padding))
            || (!header.has(flags::PADDED) && padding != 0)
        {
            return Err(Error::Protocol);
        }
        self.known_stream(header.stream)?;
        let active = self
            .streams
            .iter()
            .position(|entry| entry.id == header.stream);
        let received = active
            .map(|index| self.data_length(index, data_length, header.has(flags::END_STREAM)))
            .transpose()?;
        let written = self.credit(header.stream, header.length as u32, out)?;
        if let Some(index) = active {
            self.streams[index].received = received.unwrap();
        }
        if header.has(flags::END_STREAM) {
            self.close(header.stream);
        }
        Ok(written)
    }

    fn known_stream(&self, stream: u32) -> Result<(), Error> {
        // Push is disabled; only odd streams actually opened by us can exist.
        if stream & 1 == 0 || stream >= self.next_stream {
            return Err(Error::Protocol);
        }
        Ok(())
    }

    fn data_length(&self, index: usize, length: usize, end_stream: bool) -> Result<u64, Error> {
        let entry = &self.streams[index];
        if !entry.response_started || (entry.no_body && length != 0) {
            return Err(Error::Protocol);
        }
        let received = entry
            .received
            .checked_add(length as u64)
            .ok_or(Error::Protocol)?;
        if entry
            .content_length
            .is_some_and(|expected| received > expected || (end_stream && received != expected))
        {
            return Err(Error::Protocol);
        }
        Ok(received)
    }

    /// Return flow-control credit for `length` bytes consumed on `stream`.
    fn credit(&mut self, stream: u32, length: u32, out: &mut [u8]) -> Result<usize, Error> {
        if length == 0 {
            return Ok(0);
        }
        const THRESHOLD: u32 = RECEIVE_WINDOW / 2;
        let consumed = self.consumed.checked_add(length).ok_or(Error::Protocol)?;
        let stream_consumed = self
            .streams
            .iter()
            .find(|s| s.id == stream)
            .map(|entry| entry.consumed.checked_add(length).ok_or(Error::Protocol))
            .transpose()?;
        let connection_update = consumed >= THRESHOLD;
        let stream_update = stream_consumed.is_some_and(|value| value >= THRESHOLD);
        let required =
            (connection_update as usize + stream_update as usize) * (frame::HEADER_LEN + 4);
        if out.len() < required {
            return Err(Error::BufferTooSmall);
        }
        let mut written = 0;
        if connection_update {
            written +=
                frame::write_frame(FrameType::WindowUpdate, 0, 0, &consumed.to_be_bytes(), out)?;
        }
        if stream_update {
            written += frame::write_frame(
                FrameType::WindowUpdate,
                0,
                stream,
                &stream_consumed.unwrap().to_be_bytes(),
                &mut out[written..],
            )?;
        }
        self.consumed = if connection_update { 0 } else { consumed };
        if let Some(entry) = self.streams.iter_mut().find(|s| s.id == stream) {
            entry.consumed = if stream_update {
                0
            } else {
                stream_consumed.unwrap()
            };
        }
        Ok(written)
    }

    fn apply_settings(&mut self, payload: &[u8]) -> Result<(), Error> {
        if !payload.len().is_multiple_of(6) {
            return Err(Error::Protocol);
        }
        for entry in payload.as_chunks::<6>().0 {
            let identifier = u16::from_be_bytes([entry[0], entry[1]]);
            let value = u32::from_be_bytes([entry[2], entry[3], entry[4], entry[5]]);
            match identifier {
                settings::MAX_FRAME_SIZE => {
                    if !(16_384..=16_777_215).contains(&value) {
                        return Err(Error::Protocol);
                    }
                    self.peer_max_frame = value as usize;
                }
                settings::INITIAL_WINDOW_SIZE => {
                    if value > 0x7fff_ffff {
                        return Err(Error::Protocol);
                    }
                    let delta = value as i64 - self.peer_initial_window as i64;
                    if self
                        .streams
                        .iter()
                        .any(|stream| stream.send_window + delta > 0x7fff_ffff)
                    {
                        return Err(Error::Protocol);
                    }
                    for stream in &mut self.streams {
                        stream.send_window += delta;
                    }
                    self.peer_initial_window = value;
                }
                settings::MAX_CONCURRENT_STREAMS => self.peer_max_streams = value,
                settings::ENABLE_PUSH => return Err(Error::Protocol),
                // This limits our encoder, which deliberately has no dynamic
                // table. The receive decoder is governed by the 256-byte limit
                // we advertised and must retain its connection-wide state.
                settings::HEADER_TABLE_SIZE => {}
                // Unknown settings are ignored as required by HTTP/2.
                _ => {}
            }
        }
        Ok(())
    }

    fn close(&mut self, stream: u32) {
        if let Some(entry) = self.streams.iter_mut().find(|s| s.id == stream) {
            entry.open = false;
        }
        self.streams.retain(|s| s.open);
    }

    pub fn open_streams(&self) -> usize {
        self.streams.len()
    }
}

fn write_setting(out: &mut [u8], identifier: u16, value: u32) {
    out[0..2].copy_from_slice(&identifier.to_be_bytes());
    out[2..6].copy_from_slice(&value.to_be_bytes());
}

fn read_u32(bytes: &[u8]) -> Option<u32> {
    Some(u32::from_be_bytes(bytes.get(..4)?.try_into().ok()?))
}

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

    /// Build a frame the way a server would, for feeding to `recv`.
    fn frame_bytes(
        kind: FrameType,
        flags: u8,
        stream: u32,
        payload: &[u8],
    ) -> heapless::Vec<u8, 512> {
        let mut out = heapless::Vec::<u8, 512>::new();
        out.resize_default(512).unwrap();
        let len = frame::write_frame(kind, flags, stream, payload, &mut out).unwrap();
        out.truncate(len);
        out
    }

    fn open_connection() -> Connection {
        let mut connection = Connection::new();
        connection
            .request("GET", "/", "h", "https", &[], b"", &mut [0; 256])
            .unwrap();
        connection
    }

    fn response_connection() -> Connection {
        let mut connection = open_connection();
        let headers = frame_bytes(FrameType::Headers, flags::END_HEADERS, 1, &[0x88]);
        connection.recv(&headers, |_, _| {}, &mut [0; 64]).unwrap();
        connection
    }

    #[test]
    fn the_preface_is_exactly_what_the_rfc_requires() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let len = connection.start(&mut out).unwrap();
        assert!(out[..len].starts_with(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"));

        // Then SETTINGS, then the connection-level window update. Without the
        // last one a streaming response stops dead at 65535 bytes.
        let after = &out[frame::CLIENT_PREFACE.len()..len];
        let settings = FrameHeader::parse(after).unwrap();
        assert_eq!(settings.kind, FrameType::Settings);
        assert_eq!(settings.stream, 0);

        let update = &after[frame::HEADER_LEN + settings.length..];
        let update_header = FrameHeader::parse(update).unwrap();
        assert_eq!(update_header.kind, FrameType::WindowUpdate);
        assert_eq!(update_header.stream, 0);
        let increment = read_u32(&update[frame::HEADER_LEN..]).unwrap();
        assert_eq!(increment, RECEIVE_WINDOW - DEFAULT_WINDOW);
    }

    #[test]
    fn a_request_opens_an_odd_numbered_stream_and_ends_it() {
        let mut connection = Connection::new();
        let mut out = [0u8; 512];
        let (stream, len) = connection
            .request(
                "POST",
                "/machine/register",
                "127.0.0.1:8080",
                "http",
                &[("content-type", "application/json")],
                b"{}",
                &mut out,
            )
            .unwrap();

        // Client streams are odd, and the first is 1.
        assert_eq!(stream, 1);
        let headers = FrameHeader::parse(&out).unwrap();
        assert_eq!(headers.kind, FrameType::Headers);
        assert!(headers.has(flags::END_HEADERS));
        assert!(!headers.has(flags::END_STREAM), "a body follows");

        let data = FrameHeader::parse(&out[frame::HEADER_LEN + headers.length..]).unwrap();
        assert_eq!(data.kind, FrameType::Data);
        assert!(data.has(flags::END_STREAM));
        assert_eq!(len, frame::HEADER_LEN * 2 + headers.length + data.length);

        // The next request must not reuse the identifier.
        let (second, _) = connection
            .request("GET", "/", "h", "http", &[], b"", &mut out)
            .unwrap();
        assert_eq!(second, 3);
    }

    #[test]
    fn settings_are_acknowledged_and_an_ack_is_not() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];

        let settings = frame_bytes(FrameType::Settings, 0, 0, &[0, 5, 0, 0, 0x40, 0]);
        let (event, written) = connection.recv(&settings, |_, _| {}, &mut out).unwrap();
        assert_eq!(event, Event::Nothing);
        let ack = FrameHeader::parse(&out[..written]).unwrap();
        assert_eq!(ack.kind, FrameType::Settings);
        assert!(ack.has(flags::ACK));
        assert_eq!(ack.length, 0);

        // Acknowledging an acknowledgement would loop forever.
        let their_ack = frame_bytes(FrameType::Settings, flags::ACK, 0, &[]);
        let (_, written) = connection.recv(&their_ack, |_, _| {}, &mut out).unwrap();
        assert_eq!(written, 0);
    }

    #[test]
    fn peer_encoder_settings_do_not_reset_receive_hpack_state() {
        let mut connection = Connection::new();
        connection
            .decoder
            .decode(&[0x40, 0x01, b'x', 0x01, b'y'], |_, _| {})
            .unwrap();
        assert_eq!(connection.decoder.table().len(), 1);

        let settings = frame_bytes(FrameType::Settings, 0, 0, &[0, 1, 0, 0, 0x10, 0]);
        let mut out = [0u8; 64];
        connection.recv(&settings, |_, _| {}, &mut out).unwrap();
        assert_eq!(connection.decoder.table().len(), 1);
        assert_eq!(
            connection.decoder.table().capacity(),
            hpack::DEFAULT_TABLE_SIZE
        );
    }

    #[test]
    fn a_ping_is_echoed_exactly() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let payload = [1, 2, 3, 4, 5, 6, 7, 8];
        let ping = frame_bytes(FrameType::Ping, 0, 0, &payload);
        let (_, written) = connection.recv(&ping, |_, _| {}, &mut out).unwrap();
        let header = FrameHeader::parse(&out[..written]).unwrap();
        assert!(header.has(flags::ACK));
        assert_eq!(&out[frame::HEADER_LEN..written], &payload);
    }

    #[test]
    fn a_header_block_split_across_continuation_frames_is_reassembled() {
        // Decoding either half alone would corrupt the HPACK table for the rest
        // of the connection, so the partial frame must produce no event at all.
        let mut connection = open_connection();
        let mut out = [0u8; 256];

        // A status followed by an ordinary field, split between frames.
        let first = frame_bytes(FrameType::Headers, 0, 1, &[0x88]);
        let (event, _) = connection.recv(&first, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            event,
            Event::Nothing,
            "an unterminated block yields nothing"
        );

        let mut seen = heapless::Vec::<(heapless::String<32>, heapless::String<32>), 4>::new();
        let second = frame_bytes(
            FrameType::Continuation,
            flags::END_HEADERS,
            1,
            &[0, 1, b'x', 1, b'y'],
        );
        let (event, _) = connection
            .recv(
                &second,
                |name, value| {
                    let mut n = heapless::String::new();
                    let mut v = heapless::String::new();
                    n.push_str(name).unwrap();
                    v.push_str(value).unwrap();
                    seen.push((n, v)).unwrap();
                },
                &mut out,
            )
            .unwrap();

        assert!(matches!(event, Event::Headers { stream: 1, .. }));
        assert_eq!(seen.len(), 2);
        assert_eq!(seen[0].0.as_str(), ":status");
        assert_eq!(seen[0].1.as_str(), "200");
        assert_eq!(seen[1].0.as_str(), "x");
    }

    #[test]
    fn data_returns_the_payload_and_eventually_a_window_update() {
        let mut connection = response_connection();
        let mut out = [0u8; 256];

        let data = frame_bytes(FrameType::Data, 0, 1, b"hello");
        let (event, written) = connection.recv(&data, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            event,
            Event::Data {
                stream: 1,
                data: b"hello",
                end_stream: false
            }
        );
        // Below the threshold, so no update yet: one update per small frame
        // would be pure overhead.
        assert_eq!(written, 0);
    }

    #[test]
    fn a_long_response_gets_its_window_topped_up_before_it_can_stall() {
        // The regression test for the failure mode that looks like a server
        // fault: without this the peer stops after 65535 bytes.
        let mut connection = response_connection();
        let mut out = [0u8; 256];
        let payload = [0u8; 400];

        let mut updates = 0;
        let mut delivered = 0usize;
        // Well past both the 65535-byte default window and our own top-up
        // threshold, so a connection that never replenishes would have stalled
        // long before the end of this loop.
        let frames = 2_000;
        for _ in 0..frames {
            let data = frame_bytes(FrameType::Data, 0, 1, &payload);
            let (event, written) = connection.recv(&data, |_, _| {}, &mut out).unwrap();
            if let Event::Data { data, .. } = event {
                delivered += data.len();
            }
            if written > 0 {
                updates += 1;
                let header = FrameHeader::parse(&out[..written]).unwrap();
                assert_eq!(header.kind, FrameType::WindowUpdate);
            }
        }
        assert_eq!(delivered, frames * 400);
        assert!(
            updates > 0,
            "the connection window must be replenished, or the server stops at 65535 bytes"
        );
    }

    #[test]
    fn padding_and_priority_are_stripped_before_hpack_sees_the_block() {
        let mut connection = open_connection();
        let mut out = [0u8; 256];
        // pad length 2, priority (5 bytes), the block, then the padding.
        let payload = [2, 0, 0, 0, 0, 0, 0x88, 0xaa, 0xbb];
        let headers = frame_bytes(
            FrameType::Headers,
            flags::END_HEADERS | flags::PADDED | flags::PRIORITY,
            1,
            &payload,
        );
        let mut status = heapless::String::<8>::new();
        connection
            .recv(
                &headers,
                |name, value| {
                    if name == ":status" {
                        status.push_str(value).unwrap();
                    }
                },
                &mut out,
            )
            .unwrap();
        assert_eq!(status.as_str(), "200");
    }

    #[test]
    fn goaway_and_reset_are_reported_rather_than_hidden() {
        let mut connection = open_connection();
        let mut out = [0u8; 128];

        let reset = frame_bytes(FrameType::RstStream, 0, 1, &[0, 0, 0, 8]);
        let (event, _) = connection.recv(&reset, |_, _| {}, &mut out).unwrap();
        assert_eq!(event, Event::Reset { stream: 1, code: 8 });

        let goaway = frame_bytes(FrameType::GoAway, 0, 0, &[0, 0, 0, 1, 0, 0, 0, 2]);
        let (event, _) = connection.recv(&goaway, |_, _| {}, &mut out).unwrap();
        assert_eq!(event, Event::GoAway { code: 2 });
    }

    #[test]
    fn a_promised_push_is_a_protocol_error_because_we_refused_push() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let promise = frame_bytes(FrameType::PushPromise, flags::END_HEADERS, 1, &[0, 0, 0, 2]);
        assert_eq!(
            connection.recv(&promise, |_, _| {}, &mut out).err(),
            Some(Error::Protocol)
        );
    }

    #[test]
    fn an_unknown_frame_type_is_discarded_not_fatal() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let unknown = frame_bytes(FrameType::Unknown(0x63), 0, 0, b"whatever");
        let (event, written) = connection.recv(&unknown, |_, _| {}, &mut out).unwrap();
        assert_eq!(event, Event::Nothing);
        assert_eq!(written, 0);
    }
    #[test]
    fn failed_requests_do_not_consume_stream_slots_or_identifiers() {
        let mut connection = Connection::new();
        for _ in 0..MAX_STREAMS + 1 {
            assert_eq!(
                connection.request("GET", "/", "h", "https", &[], b"", &mut []),
                Err(Error::BufferTooSmall)
            );
            assert_eq!(connection.open_streams(), 0);
        }
        let mut out = [0u8; 128];
        assert_eq!(
            connection
                .request("GET", "/", "h", "https", &[], b"", &mut out)
                .unwrap()
                .0,
            1
        );
    }

    #[test]
    fn continuation_must_follow_headers_on_the_same_stream() {
        for (kind, stream) in [
            (FrameType::Continuation, 3),
            (FrameType::Ping, 0),
            (FrameType::Headers, 3),
        ] {
            let mut connection = open_connection();
            let mut out = [0u8; 128];
            let first = frame_bytes(FrameType::Headers, 0, 1, &[0x88]);
            connection.recv(&first, |_, _| {}, &mut out).unwrap();
            let next = frame_bytes(kind, flags::END_HEADERS, stream, &[0x82]);
            assert_eq!(
                connection.recv(&next, |_, _| {}, &mut out).err(),
                Some(Error::Protocol)
            );
        }
        let mut connection = open_connection();
        let orphan = frame_bytes(FrameType::Continuation, flags::END_HEADERS, 1, &[0x88]);
        assert_eq!(
            connection.recv(&orphan, |_, _| {}, &mut [0u8; 128]).err(),
            Some(Error::Protocol)
        );
    }
    #[test]
    fn request_body_waits_for_connection_and_stream_credit() {
        let mut connection = Connection::new();
        let mut out = [0u8; 256];
        let zero = frame_bytes(FrameType::Settings, 0, 0, &[0, 4, 0, 0, 0, 0]);
        connection.recv(&zero, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            connection.request("POST", "/", "h", "https", &[], b"a", &mut out),
            Err(Error::FlowControl)
        );
        assert_eq!(connection.open_streams(), 0);
        let one = frame_bytes(FrameType::Settings, 0, 0, &[0, 4, 0, 0, 0, 1]);
        connection.recv(&one, |_, _| {}, &mut out).unwrap();
        connection.send_window = 0;
        assert_eq!(
            connection.request("POST", "/", "h", "https", &[], b"a", &mut out),
            Err(Error::FlowControl)
        );
        let update = frame_bytes(FrameType::WindowUpdate, 0, 0, &[0, 0, 0, 1]);
        connection.recv(&update, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            connection
                .request("POST", "/", "h", "https", &[], b"a", &mut out)
                .unwrap()
                .0,
            1
        );
        assert_eq!(connection.send_window, 0);
    }

    #[test]
    fn exhausted_connection_credit_blocks_later_small_requests() {
        let mut connection = Connection::new();
        let mut out = [0u8; 17_000];
        for _ in 0..4 {
            let (id, _) = connection
                .request("POST", "/", "h", "https", &[], &[0; 16_000], &mut out)
                .unwrap();
            connection.close(id);
        }
        assert_eq!(
            connection.request("POST", "/", "h", "https", &[], &[0; 2_000], &mut out),
            Err(Error::FlowControl)
        );
    }

    #[test]
    fn failed_credit_output_can_be_retried_without_double_counting() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let (id, _) = connection
            .request("GET", "/", "h", "https", &[], b"", &mut out)
            .unwrap();
        let headers = frame_bytes(FrameType::Headers, flags::END_HEADERS, id, &[0x88]);
        connection.recv(&headers, |_, _| {}, &mut out).unwrap();
        connection.consumed = RECEIVE_WINDOW / 2 - 1;
        connection.streams[0].consumed = RECEIVE_WINDOW / 2 - 1;
        let data = frame_bytes(FrameType::Data, 0, id, b"a");
        assert_eq!(
            connection.recv(&data, |_, _| {}, &mut out[..13]).err(),
            Some(Error::BufferTooSmall)
        );
        assert_eq!(connection.consumed, RECEIVE_WINDOW / 2 - 1);
        let (_, written) = connection.recv(&data, |_, _| {}, &mut out).unwrap();
        assert_eq!(written, 26);
        assert_eq!(read_u32(&out[9..13]), Some(RECEIVE_WINDOW / 2));
        assert_eq!(read_u32(&out[22..26]), Some(RECEIVE_WINDOW / 2));
    }

    #[test]
    fn settings_stream_limits_and_goaway_stop_new_requests() {
        let mut connection = Connection::new();
        let mut out = [0u8; 128];
        let zero = frame_bytes(FrameType::Settings, 0, 0, &[0, 3, 0, 0, 0, 0]);
        connection.recv(&zero, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            connection.request("GET", "/", "h", "https", &[], b"", &mut out),
            Err(Error::TooManyStreams)
        );
        let goaway = frame_bytes(FrameType::GoAway, 0, 0, &[0; 8]);
        connection.recv(&goaway, |_, _| {}, &mut out).unwrap();
        assert_eq!(
            connection.request("GET", "/", "h", "https", &[], b"", &mut out),
            Err(Error::GoAway)
        );
    }

    #[test]
    fn malformed_control_frames_are_rejected() {
        for (kind, flags, stream, payload) in [
            (FrameType::Settings, flags::ACK, 0, &[0u8; 6][..]),
            (FrameType::Settings, 0, 1, &[][..]),
            (FrameType::Ping, 0, 0, &[0; 7][..]),
            (FrameType::RstStream, 0, 0, &[0; 4][..]),
            (FrameType::WindowUpdate, 0, 0, &[0; 4][..]),
            (FrameType::Headers, flags::END_HEADERS, 0, &[0x88][..]),
        ] {
            let bytes = frame_bytes(kind, flags, stream, payload);
            assert_eq!(
                Connection::new()
                    .recv(&bytes, |_, _| {}, &mut [0u8; 128])
                    .err(),
                Some(Error::Protocol)
            );
        }
    }
}