tailfeather 0.2.1

//! Transport-independent ts2021 Noise and HTTP/2 control sessions.
//!
//! TCP, DNS, and TLS stay in the platform adapter. Once an adapter yields an
//! [`embedded_io_async`] stream, this module performs the same `/key`, Noise
//! upgrade, registration, streaming-map, and endpoint-update protocol on a
//! workstation or an Embassy target.

use core::borrow::BorrowMut;

use embedded_io_async::{Read, Write};
use micro_h2::{Connection, Event, frame};
use x25519_dalek::StaticSecret;

use crate::control::CAPABILITY_VERSION;
use crate::keys::{Identity, MachinePublic, Rng};
use crate::netmap::{FrameReader, Netmap, Parser};
use crate::noise::{self, Session, upgrade};

const HTTP_BUFFER: usize = 2048;
// Control frames and response headers are small. DATA payloads are streamed
// through a separate record-sized plaintext window rather than retained as a
// complete 16 KiB HTTP/2 frame.
pub const H2_BUFFER: usize = 2 * 1024;
const H2_OUT_BUFFER: usize = 4096;
const H2_STATUS_SLOTS: usize = 8;

#[derive(Debug, Clone, Copy)]
pub enum Error<E> {
    Io(E),
    UnexpectedEof,
    Noise(noise::Error),
    Http2(micro_h2::Error),
    Netmap(crate::netmap::Error),
    NoServerKey,
    Key(crate::keys::DecodeError),
    UnexpectedStatus(u16),
    ResponseTooLong,
    FrameTooLarge,
    InterleavedStream(u32),
    StreamEnded,
}

impl<E: core::fmt::Display> core::fmt::Display for Error<E> {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::Io(error) => write!(f, "control I/O: {error}"),
            Self::UnexpectedEof => f.write_str("control stream ended mid-message"),
            Self::Noise(error) => write!(f, "ts2021: {error}"),
            Self::Http2(error) => write!(f, "HTTP/2: {error}"),
            Self::Netmap(error) => write!(f, "netmap: {error}"),
            Self::NoServerKey => f.write_str("control server published no machine key"),
            Self::Key(error) => write!(f, "control key: {error}"),
            Self::UnexpectedStatus(status) => write!(f, "control server answered {status}"),
            Self::ResponseTooLong => f.write_str("control response exceeds its static buffer"),
            Self::FrameTooLarge => f.write_str("HTTP/2 frame exceeds its static buffer"),
            Self::InterleavedStream(stream) => {
                write!(f, "HTTP/2 stream {stream} requires its own consumer")
            }
            Self::StreamEnded => f.write_str("streaming map ended"),
        }
    }
}

impl<E: core::fmt::Debug + core::fmt::Display> core::error::Error for Error<E> {}

fn read_error<E>(error: embedded_io_async::ReadExactError<E>) -> Error<E> {
    match error {
        embedded_io_async::ReadExactError::UnexpectedEof => Error::UnexpectedEof,
        embedded_io_async::ReadExactError::Other(error) => Error::Io(error),
    }
}

async fn read_exact_progress<S>(
    stream: &mut S,
    output: &mut [u8],
    filled: &mut u16,
) -> Result<(), Error<S::Error>>
where
    S: Read,
{
    while usize::from(*filled) < output.len() {
        let start = usize::from(*filled);
        let amount = stream.read(&mut output[start..]).await.map_err(Error::Io)?;
        if amount == 0 {
            return Err(Error::UnexpectedEof);
        }
        *filled += amount as u16;
    }
    Ok(())
}

/// Read `GET /key` from a connection the platform has already authenticated.
pub async fn fetch_server_key<S>(
    stream: &mut S,
    authority: &str,
) -> Result<MachinePublic, Error<S::Error>>
where
    S: Read + Write,
{
    let mut request = [0; upgrade::MAX_REQUEST];
    let request = upgrade::write_key_request(authority, &mut request).map_err(Error::Noise)?;
    stream.write_all(request).await.map_err(Error::Io)?;
    stream.flush().await.map_err(Error::Io)?;

    let mut response = [0; HTTP_BUFFER];
    let mut filled = 0;
    loop {
        if filled == response.len() {
            return Err(Error::ResponseTooLong);
        }
        if let Some(complete) = response_length(&response[..filled]).map_err(Error::Noise)? {
            if complete > response.len() {
                return Err(Error::ResponseTooLong);
            }
            if filled >= complete {
                break;
            }
        }
        let read = stream
            .read(&mut response[filled..])
            .await
            .map_err(Error::Io)?;
        if read == 0 {
            break;
        }
        filled += read;
    }
    let parsed = upgrade::parse_response(&response[..filled]).map_err(Error::Noise)?;
    if parsed.status != 200 {
        return Err(Error::UnexpectedStatus(parsed.status));
    }
    let body = &response[parsed.header_len..filled];
    let key = json_string(body, "publicKey").ok_or(Error::NoServerKey)?;
    MachinePublic::parse(key).map_err(Error::Key)
}

/// Stop at an explicit HTTP/1 body boundary instead of waiting for the peer to
/// close TLS. Some embedded TCP adapters surface a clean TLS `close_notify` or
/// the following FIN as an I/O error, even after the complete response arrived.
fn response_length(response: &[u8]) -> Result<Option<usize>, noise::Error> {
    let Some(header_end) = response.windows(4).position(|bytes| bytes == b"\r\n\r\n") else {
        return Ok(None);
    };
    let header_len = header_end + 4;
    let headers =
        core::str::from_utf8(&response[..header_end]).map_err(|_| noise::Error::Malformed)?;
    let mut content_length = None;
    for line in headers.split("\r\n").skip(1) {
        let Some((name, value)) = line.split_once(':') else {
            return Err(noise::Error::Malformed);
        };
        if name.eq_ignore_ascii_case("content-length") {
            let length = value
                .trim()
                .parse::<usize>()
                .map_err(|_| noise::Error::Malformed)?;
            if content_length.replace(length).is_some() {
                return Err(noise::Error::Malformed);
            }
        }
    }
    let Some(content_length) = content_length else {
        return Ok(None);
    };
    header_len
        .checked_add(content_length)
        .map(Some)
        .ok_or(noise::Error::Malformed)
}

/// The first ts2021 handshake message, ready to send.
///
/// Over TCP it travels base64-encoded in the `X-Tailscale-Handshake` header of
/// the `POST /ts2021` upgrade. A WebSocket client cannot set request headers,
/// so it carries the same bytes in the query string instead; either way the
/// bytes exist before the stream that will carry the response.
pub struct Initiation {
    bytes: [u8; noise::INITIATION_LEN],
    len: usize,
}

impl Initiation {
    pub fn as_bytes(&self) -> &[u8] {
        &self.bytes[..self.len]
    }
}

/// Client handshake state between [`initiate`] and [`complete`].
pub struct PendingHandshake {
    handshake: noise::Handshake,
}

/// Start a ts2021 handshake without touching any stream.
///
/// This is the half of [`establish`] a transport needs when the initiation must
/// be known before the connection is opened, as with ts2021 over WebSocket.
pub fn initiate(
    identity: &Identity,
    server_key: &MachinePublic,
    rng: &mut impl Rng,
) -> Result<(Initiation, PendingHandshake), noise::Error> {
    let mut ephemeral = [0; 32];
    rng.fill_bytes(&mut ephemeral);
    let ephemeral = StaticSecret::from(ephemeral);
    let mut bytes = [0; noise::INITIATION_LEN];
    let (len, handshake) = noise::initiate(
        identity.machine.secret(),
        server_key.as_public(),
        &ephemeral,
        &mut bytes,
    )?;
    Ok((Initiation { bytes, len }, PendingHandshake { handshake }))
}

/// Finish a handshake on a stream whose next bytes are the server's Noise
/// response, with no HTTP/1.1 exchange in front of it.
///
/// This is the WebSocket shape. There the server may say nothing more until
/// the client's HTTP/2 preface arrives, so the optional early payload is read
/// only once HTTP/2 has started; see [`DeferredNoiseStream`].
pub async fn complete<S>(
    mut stream: S,
    pending: PendingHandshake,
) -> Result<DeferredNoiseStream<S>, Error<S::Error>>
where
    S: Read + Write,
{
    let session = read_noise_response(&mut stream, pending, &[]).await?;
    Ok(DeferredNoiseStream { stream, session })
}

/// Read the server's Noise response, of which the first `carried` bytes were
/// already read from the stream, and derive the record session.
async fn read_noise_response<S>(
    stream: &mut S,
    pending: PendingHandshake,
    carried: &[u8],
) -> Result<Session, Error<S::Error>>
where
    S: Read + Write,
{
    let mut noise_response = [0; noise::RESPONSE_LEN];
    let carried = &carried[..carried.len().min(noise_response.len())];
    noise_response[..carried.len()].copy_from_slice(carried);
    if carried.len() < noise_response.len() {
        stream
            .read_exact(&mut noise_response[carried.len()..])
            .await
            .map_err(read_error)?;
    }
    pending
        .handshake
        .consume_response(&noise_response)
        .map_err(Error::Noise)
}

/// Read records until the early-payload question is answered. Plaintext past
/// the payload — or the probe itself, when it was not the payload magic —
/// belongs to HTTP/2 and is left in `pushback`.
async fn read_early_payload<S>(
    stream: &mut S,
    session: &mut Session,
    pushback: &mut heapless::Vec<u8, { noise::MAX_PLAINTEXT }>,
) -> Result<(), Error<S::Error>>
where
    S: Read,
{
    let mut reader = noise::EarlyReader::new();
    let mut plaintext = [0; noise::MAX_PLAINTEXT];
    while !reader.is_complete() {
        let len = read_record(stream, session, &mut plaintext).await?;
        let consumed = reader.push(&plaintext[..len]).map_err(Error::Noise)?;
        pushback
            .extend_from_slice(&plaintext[consumed..len])
            .map_err(|_| Error::ResponseTooLong)?;
    }
    if let noise::early::Early::PushBack(bytes) = reader.finish().map_err(Error::Noise)? {
        let mut carried = heapless::Vec::new();
        carried
            .extend_from_slice(bytes)
            .map_err(|_| Error::ResponseTooLong)?;
        carried
            .extend_from_slice(pushback)
            .map_err(|_| Error::ResponseTooLong)?;
        *pushback = carried;
    }
    Ok(())
}

async fn read_record<S>(
    stream: &mut S,
    session: &mut Session,
    out: &mut [u8],
) -> Result<usize, Error<S::Error>>
where
    S: Read,
{
    let mut header = [0; noise::HEADER_LEN];
    stream.read_exact(&mut header).await.map_err(read_error)?;
    let len = Session::ciphertext_len(&header).map_err(Error::Noise)?;
    let mut ciphertext = [0; noise::MAX_MESSAGE];
    stream
        .read_exact(&mut ciphertext[..len])
        .await
        .map_err(read_error)?;
    session.open(&ciphertext[..len], out).map_err(Error::Noise)
}

/// Upgrade a fresh authenticated stream to ts2021 and consume its optional
/// early payload. The returned stream is ready for one reusable HTTP/2 session.
///
/// Composed of [`initiate`] and the same Noise-response handling as
/// [`complete`] around the HTTP/1.1 upgrade. Over HTTP/1.1 the server sends
/// its early payload straight away, so it is consumed here.
pub async fn establish<S>(
    mut stream: S,
    authority: &str,
    identity: &Identity,
    server_key: &MachinePublic,
    rng: &mut impl Rng,
) -> Result<NoiseStream<S>, Error<S::Error>>
where
    S: Read + Write,
{
    let (initiation, pending) = initiate(identity, server_key, rng).map_err(Error::Noise)?;
    let mut request = [0; upgrade::MAX_REQUEST];
    let request = upgrade::write_upgrade_request(authority, initiation.as_bytes(), &mut request)
        .map_err(Error::Noise)?;
    stream.write_all(request).await.map_err(Error::Io)?;
    stream.flush().await.map_err(Error::Io)?;

    let mut response = [0; HTTP_BUFFER];
    let mut filled = 0;
    let parsed = loop {
        match upgrade::parse_response(&response[..filled]) {
            Ok(parsed) => break parsed,
            Err(noise::Error::Incomplete) => {}
            Err(error) => return Err(Error::Noise(error)),
        }
        if filled == response.len() {
            return Err(Error::ResponseTooLong);
        }
        // Stop exactly at the header boundary. A larger speculative read would
        // require retaining up to another 2 KiB of encrypted Noise bytes for
        // the lifetime of the HTTP/2 connection.
        let read = stream
            .read(&mut response[filled..filled + 1])
            .await
            .map_err(Error::Io)?;
        if read == 0 {
            return Err(Error::UnexpectedEof);
        }
        filled += read;
    };
    if !upgrade::is_upgrade(&response[..filled], parsed) {
        return Err(Error::UnexpectedStatus(parsed.status));
    }
    let session =
        read_noise_response(&mut stream, pending, &response[parsed.header_len..filled]).await?;
    let mut stream = NoiseStream {
        stream,
        session,
        pushback: heapless::Vec::new(),
    };
    stream.read_early().await?;
    Ok(stream)
}

pub struct NoiseStream<S> {
    stream: S,
    session: Session,
    pushback: heapless::Vec<u8, { noise::MAX_PLAINTEXT }>,
}

impl<S> NoiseStream<S>
where
    S: Read + Write,
{
    async fn read_early(&mut self) -> Result<(), Error<S::Error>> {
        read_early_payload(&mut self.stream, &mut self.session, &mut self.pushback).await
    }

    pub async fn into_http2(self) -> Result<Http2<S>, Error<S::Error>> {
        Http2::start_carried(self.stream, self.session, &self.pushback, false).await
    }

    /// Start HTTP/2 with an HTTPS pseudo-scheme after a verified TLS adapter.
    pub async fn into_https2(self) -> Result<Http2<S>, Error<S::Error>> {
        Http2::start_carried(self.stream, self.session, &self.pushback, true).await
    }

    /// Start HTTPS-flavoured HTTP/2 directly in caller-owned storage.
    ///
    /// Embedded executors otherwise reserve space for both the returned
    /// connection and the future receiving it. Placing the connection before
    /// the first HTTP/2 write keeps exactly one copy alive.
    pub async fn into_https2_placed<'a>(
        self,
        place: impl FnOnce(Http2<S>) -> &'a mut Http2<S>,
    ) -> Result<&'a mut Http2<S>, Error<S::Error>>
    where
        S: 'a,
    {
        Http2::start_carried_placed(self.stream, self.session, &self.pushback, true, place).await
    }

    /// Start plaintext-flavoured HTTP/2 directly in caller-owned storage.
    pub async fn into_http2_placed<'a>(
        self,
        place: impl FnOnce(Http2<S>) -> &'a mut Http2<S>,
    ) -> Result<&'a mut Http2<S>, Error<S::Error>>
    where
        S: 'a,
    {
        Http2::start_carried_placed(self.stream, self.session, &self.pushback, false, place).await
    }

    pub async fn into_https2_with_buffer(
        self,
        buffer: &mut [u8; H2_BUFFER],
    ) -> Result<Http2<S, &mut [u8; H2_BUFFER]>, Error<S::Error>> {
        Http2::start_with_storage(
            self.stream,
            self.session,
            &self.pushback,
            true,
            buffer,
            Connection::new(),
        )
        .await
    }

    /// Start HTTPS-flavoured HTTP/2 using caller-owned receive and protocol state.
    /// The connection state is reset for each session, allowing reconnects to
    /// reuse one allocation-free slot without copying it through async futures.
    pub async fn into_https2_with_storage<'a>(
        self,
        buffer: &'a mut [u8; H2_BUFFER],
        connection: &'a mut Connection,
    ) -> Result<Http2<S, &'a mut [u8; H2_BUFFER], &'a mut Connection>, Error<S::Error>> {
        self.into_carried_with_storage(buffer, connection, true)
            .await
    }

    /// Start plaintext-flavoured HTTP/2 using reusable receive and protocol state.
    pub async fn into_http2_with_storage<'a>(
        self,
        buffer: &'a mut [u8; H2_BUFFER],
        connection: &'a mut Connection,
    ) -> Result<Http2<S, &'a mut [u8; H2_BUFFER], &'a mut Connection>, Error<S::Error>> {
        self.into_carried_with_storage(buffer, connection, false)
            .await
    }

    async fn into_carried_with_storage<'a>(
        self,
        buffer: &'a mut [u8; H2_BUFFER],
        connection: &'a mut Connection,
        secure: bool,
    ) -> Result<Http2<S, &'a mut [u8; H2_BUFFER], &'a mut Connection>, Error<S::Error>> {
        *connection = Connection::new();
        Http2::start_with_storage(
            self.stream,
            self.session,
            &self.pushback,
            secure,
            buffer,
            connection,
        )
        .await
    }
}

/// A ts2021 stream from [`complete`] whose optional early payload has not been
/// read yet.
///
/// Starting HTTP/2 writes the client preface first and then reads the early
/// payload, because a WebSocket-accepting server may send nothing — neither
/// the payload nor its own HTTP/2 preface — until the client has spoken.
pub struct DeferredNoiseStream<S> {
    stream: S,
    session: Session,
}

impl<S> DeferredNoiseStream<S>
where
    S: Read + Write,
{
    pub async fn into_http2(self) -> Result<Http2<S>, Error<S::Error>> {
        let mut http2 = Http2::start_carried(self.stream, self.session, &[], false).await?;
        http2.read_deferred_early().await?;
        Ok(http2)
    }

    /// Start HTTP/2 with an HTTPS pseudo-scheme, as over a `wss:` socket.
    pub async fn into_https2(self) -> Result<Http2<S>, Error<S::Error>> {
        let mut http2 = Http2::start_carried(self.stream, self.session, &[], true).await?;
        http2.read_deferred_early().await?;
        Ok(http2)
    }
}

/// One reusable HTTP/2 connection over the ts2021 record layer.
///
/// The default `B = [u8; H2_BUFFER]` owns its receive storage and remains the
/// convenient host interface. Firmware may borrow caller-owned storage through
/// [`NoiseStream::into_https2_with_buffer`] so reconnects can reuse it.
/// The `C` parameter similarly allows [`NoiseStream::into_https2_with_storage`]
/// to borrow the protocol state instead of carrying it through async futures.
pub struct Http2<S, B = [u8; H2_BUFFER], C = Connection> {
    stream: S,
    session: Session,
    connection: C,
    buffer: B,
    filled: u16,
    plaintext: [u8; noise::MAX_PLAINTEXT],
    plaintext_start: u16,
    plaintext_end: u16,
    record_scratch: [u8; noise::TAG_LEN],
    record_progress: u16,
    record_state: u16,
    data: Option<DataFrame>,
    statuses: Statuses,
    status: u16,
    secure: bool,
}

#[derive(Clone, Copy)]
struct DataFrame {
    header: frame::FrameHeader,
    remaining: u16,
    padding: u16,
}

struct Statuses {
    streams: heapless::Vec<u32, H2_STATUS_SLOTS>,
    values: [u16; H2_STATUS_SLOTS],
}

impl Statuses {
    const fn new() -> Self {
        Self {
            streams: heapless::Vec::new(),
            values: [0; H2_STATUS_SLOTS],
        }
    }

    fn get(&self, stream: u32) -> Option<u16> {
        self.streams
            .iter()
            .position(|id| *id == stream)
            .map(|index| self.values[index])
    }

    fn set(&mut self, stream: u32, status: u16) -> Result<(), ()> {
        if let Some(index) = self.streams.iter().position(|id| *id == stream) {
            self.values[index] = status;
            return Ok(());
        }
        let index = self.streams.len();
        self.streams.push(stream).map_err(|_| ())?;
        self.values[index] = status;
        Ok(())
    }

    fn take(&mut self, stream: u32) -> Option<u16> {
        let index = self.streams.iter().position(|id| *id == stream)?;
        let last = self.streams.len() - 1;
        let status = self.values[index];
        self.streams.swap_remove(index);
        if index != last {
            self.values[index] = self.values[last];
        }
        Some(status)
    }
}

enum Input<'a> {
    Frame(usize),
    Data {
        stream: u32,
        bytes: &'a [u8],
        end_stream: bool,
        frame_done: bool,
    },
}

impl<S> Http2<S>
where
    S: Read + Write,
{
    pub async fn start(
        stream: S,
        session: Session,
        pushback: &[u8],
    ) -> Result<Self, Error<S::Error>> {
        Self::start_carried(stream, session, pushback, false).await
    }

    async fn start_carried(
        stream: S,
        session: Session,
        pushback: &[u8],
        secure: bool,
    ) -> Result<Self, Error<S::Error>> {
        Self::start_with_storage(
            stream,
            session,
            pushback,
            secure,
            [0; H2_BUFFER],
            Connection::new(),
        )
        .await
    }

    async fn start_carried_placed<'a>(
        stream: S,
        session: Session,
        pushback: &[u8],
        secure: bool,
        place: impl FnOnce(Self) -> &'a mut Self,
    ) -> Result<&'a mut Self, Error<S::Error>>
    where
        S: 'a,
    {
        let (client, out, len) = Self::prepare(
            stream,
            session,
            pushback,
            secure,
            [0; H2_BUFFER],
            Connection::new(),
        )?;
        let client = place(client);
        client.send(&out[..len]).await?;
        Ok(client)
    }
}

impl<S, B, C> Http2<S, B, C>
where
    S: Read + Write,
    B: AsRef<[u8]> + AsMut<[u8]>,
    C: BorrowMut<Connection>,
{
    async fn start_with_storage(
        stream: S,
        session: Session,
        pushback: &[u8],
        secure: bool,
        buffer: B,
        connection: C,
    ) -> Result<Self, Error<S::Error>> {
        let (mut client, out, len) =
            Self::prepare(stream, session, pushback, secure, buffer, connection)?;
        client.send(&out[..len]).await?;
        Ok(client)
    }

    /// Consume an early payload that arrives after the client preface. Nothing
    /// has been read through HTTP/2 yet, so what follows the payload becomes
    /// the start of the plaintext HTTP/2 input.
    async fn read_deferred_early(&mut self) -> Result<(), Error<S::Error>> {
        let mut pushback = heapless::Vec::<u8, { noise::MAX_PLAINTEXT }>::new();
        read_early_payload(&mut self.stream, &mut self.session, &mut pushback).await?;
        self.plaintext[..pushback.len()].copy_from_slice(&pushback);
        self.plaintext_start = 0;
        self.plaintext_end = u16::try_from(pushback.len()).map_err(|_| Error::FrameTooLarge)?;
        Ok(())
    }

    fn prepare(
        stream: S,
        session: Session,
        pushback: &[u8],
        secure: bool,
        mut buffer: B,
        connection: C,
    ) -> Result<(Self, [u8; 256], usize), Error<S::Error>> {
        buffer.as_mut().fill(0);
        let mut client = Self {
            stream,
            session,
            connection,
            buffer,
            filled: 0,
            plaintext: [0; noise::MAX_PLAINTEXT],
            plaintext_start: 0,
            plaintext_end: pushback.len() as u16,
            record_scratch: [0; noise::TAG_LEN],
            record_progress: 0,
            record_state: 0,
            data: None,
            statuses: Statuses::new(),
            status: 0,
            secure,
        };
        client.plaintext[..pushback.len()].copy_from_slice(pushback);
        let mut out = [0; 256];
        let len = client
            .connection
            .borrow_mut()
            .start(&mut out)
            .map_err(Error::Http2)?;
        Ok((client, out, len))
    }

    pub async fn request(
        &mut self,
        method: &str,
        path: &str,
        authority: &str,
        request_body: &[u8],
        response: &mut [u8],
    ) -> Result<(u16, usize), Error<S::Error>> {
        let stream = self
            .send_request(method, path, authority, request_body)
            .await?;
        let mut written = 0;
        self.status = 0;
        loop {
            let (len, finished) = self.read_chunk(stream, &mut response[written..]).await?;
            written += len;
            if finished {
                self.status = self.take_status(stream).unwrap_or(0);
                return Ok((self.status, written));
            }
        }
    }

    pub async fn send_request(
        &mut self,
        method: &str,
        path: &str,
        authority: &str,
        request_body: &[u8],
    ) -> Result<u32, Error<S::Error>> {
        let mut out = [0; H2_OUT_BUFFER];
        let (stream, len) = self
            .connection
            .borrow_mut()
            .request(
                method,
                path,
                authority,
                if self.secure { "https" } else { "http" },
                &[("content-type", "application/json")],
                request_body,
                &mut out,
            )
            .map_err(Error::Http2)?;
        self.send(&out[..len]).await?;
        Ok(stream)
    }

    pub async fn read_chunk(
        &mut self,
        wanted_stream: u32,
        body_out: &mut [u8],
    ) -> Result<(usize, bool), Error<S::Error>> {
        loop {
            let input = self.next_input().await?;
            if let Input::Data {
                stream,
                bytes,
                end_stream,
                frame_done,
            } = input
            {
                let len = bytes.len();
                let mut written = 0;
                let mut error = None;
                if stream == wanted_stream {
                    if len > body_out.len() {
                        error = Some(Error::FrameTooLarge);
                    } else {
                        body_out[..len].copy_from_slice(bytes);
                        written = len;
                    }
                }
                self.consume_data(len).await?;
                if let Some(error) = error {
                    return Err(error);
                }
                let finished = stream == wanted_stream && end_stream && frame_done;
                if written != 0 || finished {
                    return Ok((written, finished));
                }
                continue;
            }
            let Input::Frame(frame_len) = input else {
                unreachable!();
            };
            let mut replies = [0; 256];
            let mut parsed_status = None;
            let (event, reply_len) = self
                .connection
                .borrow_mut()
                .recv(
                    &self.buffer.as_ref()[..frame_len],
                    |name, value| {
                        if name == ":status" {
                            parsed_status = Some(value.parse().unwrap_or(0));
                        }
                    },
                    &mut replies,
                )
                .map_err(Error::Http2)?;
            let header_status = match &event {
                Event::Headers { stream, .. } => parsed_status.map(|status| (*stream, status)),
                _ => None,
            };
            let written = 0;
            let mut finished = false;
            let mut error = None;
            match event {
                Event::Headers { stream, end_stream } if stream == wanted_stream => {
                    finished = end_stream;
                }
                Event::Data { .. } => unreachable!("DATA is streamed before frame dispatch"),
                Event::Reset { stream, .. } if stream == wanted_stream => {
                    error = Some(Error::Http2(micro_h2::Error::StreamReset));
                }
                Event::GoAway { .. } => error = Some(Error::Http2(micro_h2::Error::GoAway)),
                _ => {}
            }
            if let Some((stream, status)) = header_status {
                self.set_status(stream, status)?;
            }
            self.consume(frame_len);
            if reply_len != 0 {
                self.send(&replies[..reply_len]).await?;
            }
            if let Some(error) = error {
                return Err(error);
            }
            if written != 0 || finished {
                return Ok((written, finished));
            }
        }
    }

    /// Consume a response to completion without retaining its body.
    pub async fn drain_stream(&mut self, wanted_stream: u32) -> Result<(), Error<S::Error>> {
        loop {
            let input = self.next_input().await?;
            if let Input::Data {
                stream,
                bytes,
                end_stream,
                frame_done,
            } = input
            {
                let len = bytes.len();
                self.consume_data(len).await?;
                if stream == wanted_stream && end_stream && frame_done {
                    return Ok(());
                }
                continue;
            }
            let Input::Frame(frame_len) = input else {
                unreachable!();
            };
            let header = frame::FrameHeader::parse(&self.buffer.as_ref()[..frame_len])
                .map_err(Error::Http2)?;
            if header.stream != 0 && header.stream != wanted_stream {
                return Err(Error::InterleavedStream(header.stream));
            }
            let mut replies = [0; 256];
            let mut parsed_status = None;
            let (event, reply_len) = self
                .connection
                .borrow_mut()
                .recv(
                    &self.buffer.as_ref()[..frame_len],
                    |name, value| {
                        if name == ":status" {
                            parsed_status = Some(value.parse().unwrap_or(0));
                        }
                    },
                    &mut replies,
                )
                .map_err(Error::Http2)?;
            let header_status = match &event {
                Event::Headers { stream, .. } => parsed_status.map(|status| (*stream, status)),
                _ => None,
            };
            let mut finished = false;
            let mut error = None;
            match event {
                Event::Headers { stream, end_stream } if stream == wanted_stream => {
                    finished = end_stream;
                }
                Event::Data { .. } => unreachable!("DATA is streamed before frame dispatch"),
                Event::Reset { stream, .. } if stream == wanted_stream => {
                    error = Some(Error::Http2(micro_h2::Error::StreamReset));
                }
                Event::GoAway { .. } => error = Some(Error::Http2(micro_h2::Error::GoAway)),
                _ => {}
            }
            if let Some((stream, status)) = header_status {
                self.set_status(stream, status)?;
            }
            self.consume(frame_len);
            if reply_len != 0 {
                self.send(&replies[..reply_len]).await?;
            }
            if let Some(error) = error {
                return Err(error);
            }
            if finished {
                return Ok(());
            }
        }
    }

    /// Status of the response most recently read through [`Self::read_chunk`].
    /// Connection-level frames and headers for other streams do not change it.
    pub fn status(&self) -> u16 {
        self.status
    }

    /// Status received for an in-flight response, if its headers have arrived.
    ///
    /// Streaming responses do not reach the completion point where [`Self::status`]
    /// is normally finalised, and several streams may be active at once. Callers
    /// consuming such a stream must therefore query the status by stream id.
    pub fn stream_status(&self, stream: u32) -> Option<u16> {
        self.statuses.get(stream)
    }

    /// Wait until decrypted HTTP/2 input is buffered, without consuming it.
    ///
    /// Unlike request and map operations, which also write flow-control and
    /// PING/SETTINGS acknowledgements, this only reads. Partial Noise record
    /// progress is retained in the connection, so dropping this future loses
    /// nothing provided the underlying stream's reads are cancellation-safe.
    pub async fn wait_readable(&mut self) -> Result<(), Error<S::Error>> {
        while self.filled == 0
            && self.plaintext_start == self.plaintext_end
            && self.data.is_none_or(|data| data.remaining != 0)
        {
            self.fill().await?;
        }
        Ok(())
    }

    async fn send(&mut self, mut plaintext: &[u8]) -> Result<(), Error<S::Error>> {
        let mut record = [0; noise::MAX_MESSAGE];
        while !plaintext.is_empty() {
            let take = plaintext.len().min(noise::MAX_PLAINTEXT);
            let len = self
                .session
                .seal(&plaintext[..take], &mut record)
                .map_err(Error::Noise)?;
            self.stream
                .write_all(&record[..len])
                .await
                .map_err(Error::Io)?;
            plaintext = &plaintext[take..];
        }
        self.stream.flush().await.map_err(Error::Io)
    }

    async fn next_input(&mut self) -> Result<Input<'_>, Error<S::Error>> {
        loop {
            if let Some(data) = self.data {
                if data.remaining == 0 {
                    return Ok(Input::Data {
                        stream: data.header.stream,
                        bytes: &[],
                        end_stream: data.header.has(micro_h2::frame::flags::END_STREAM),
                        frame_done: true,
                    });
                }
                if self.filled == 0 {
                    self.fill().await?;
                    continue;
                }
                let len = usize::from(self.filled.min(data.remaining));
                return Ok(Input::Data {
                    stream: data.header.stream,
                    bytes: &self.buffer.as_ref()[..len],
                    end_stream: data.header.has(micro_h2::frame::flags::END_STREAM),
                    frame_done: len == usize::from(data.remaining),
                });
            }
            if usize::from(self.filled) >= frame::HEADER_LEN {
                let header =
                    frame::FrameHeader::parse(self.buffer.as_ref()).map_err(Error::Http2)?;
                if header.length > frame::DEFAULT_MAX_FRAME {
                    return Err(Error::FrameTooLarge);
                }
                let total = frame::HEADER_LEN + header.length;
                if header.kind == micro_h2::FrameType::Data {
                    self.consume(frame::HEADER_LEN);
                    let (remaining, padding) = if header.has(micro_h2::frame::flags::PADDED) {
                        if self.filled == 0 {
                            self.fill().await?;
                        }
                        let padding = usize::from(self.buffer.as_ref()[0]);
                        self.consume(1);
                        let remaining = header
                            .length
                            .checked_sub(1 + padding)
                            .ok_or(Error::Http2(micro_h2::Error::Protocol))?;
                        (
                            u16::try_from(remaining).map_err(|_| Error::FrameTooLarge)?,
                            u16::try_from(padding).map_err(|_| Error::FrameTooLarge)?,
                        )
                    } else {
                        (
                            u16::try_from(header.length).map_err(|_| Error::FrameTooLarge)?,
                            0,
                        )
                    };
                    self.data = Some(DataFrame {
                        header,
                        remaining,
                        padding,
                    });
                    continue;
                }
                if total > self.buffer.as_ref().len() {
                    return Err(Error::FrameTooLarge);
                }
                if usize::from(self.filled) >= total {
                    return Ok(Input::Frame(total));
                }
            }
            self.fill().await?;
        }
    }

    async fn consume_data(&mut self, len: usize) -> Result<(), Error<S::Error>> {
        let mut data = self.data.ok_or(Error::Http2(micro_h2::Error::Protocol))?;
        if len > usize::from(data.remaining) || len > usize::from(self.filled) {
            return Err(Error::Http2(micro_h2::Error::Protocol));
        }
        self.consume(len);
        data.remaining -= u16::try_from(len).map_err(|_| Error::FrameTooLarge)?;
        if data.remaining != 0 {
            self.data = Some(data);
            return Ok(());
        }

        let mut padding = usize::from(data.padding);
        while padding != 0 {
            if self.filled == 0 {
                self.fill().await?;
            }
            let take = padding.min(usize::from(self.filled));
            self.consume(take);
            padding -= take;
        }
        self.data = None;
        let mut reply = [0; 2 * frame::HEADER_LEN + 8];
        let reply_len = self
            .connection
            .borrow_mut()
            .finish_data_with_length(
                data.header,
                data.header.length
                    - usize::from(data.padding)
                    - usize::from(data.header.has(micro_h2::frame::flags::PADDED)),
                &mut reply,
            )
            .map_err(Error::Http2)?;
        if reply_len != 0 {
            self.send(&reply[..reply_len]).await?;
        }
        Ok(())
    }

    async fn fill(&mut self) -> Result<(), Error<S::Error>> {
        const TAG_PHASE: u16 = 1 << 15;

        if self.plaintext_start != self.plaintext_end {
            self.copy_plaintext()?;
            return Ok(());
        }
        if self.record_state == 0 {
            read_exact_progress(
                &mut self.stream,
                &mut self.record_scratch[..noise::HEADER_LEN],
                &mut self.record_progress,
            )
            .await?;
            let ciphertext_len =
                Session::ciphertext_len(&self.record_scratch).map_err(Error::Noise)?;
            let plaintext_len = ciphertext_len - noise::TAG_LEN;
            self.record_state =
                u16::try_from(plaintext_len + 1).map_err(|_| Error::FrameTooLarge)?;
            self.record_progress = 0;
        }
        let plaintext_len = usize::from((self.record_state & !TAG_PHASE) - 1);
        if self.record_state & TAG_PHASE == 0 {
            read_exact_progress(
                &mut self.stream,
                &mut self.plaintext[..plaintext_len],
                &mut self.record_progress,
            )
            .await?;
            self.record_state |= TAG_PHASE;
            self.record_progress = 0;
        }
        read_exact_progress(
            &mut self.stream,
            &mut self.record_scratch[..noise::TAG_LEN],
            &mut self.record_progress,
        )
        .await?;
        let plaintext_len = self
            .session
            .open_parts(
                &mut self.plaintext[..plaintext_len],
                &self.record_scratch[..noise::TAG_LEN],
            )
            .map_err(Error::Noise)?;
        self.record_progress = 0;
        self.record_state = 0;
        self.plaintext_start = 0;
        self.plaintext_end = u16::try_from(plaintext_len).map_err(|_| Error::FrameTooLarge)?;
        self.copy_plaintext()?;
        Ok(())
    }

    fn copy_plaintext(&mut self) -> Result<(), Error<S::Error>> {
        let filled = usize::from(self.filled);
        let start = usize::from(self.plaintext_start);
        let end = usize::from(self.plaintext_end);
        let available = self.buffer.as_ref().len() - filled;
        if available == 0 {
            return Err(Error::FrameTooLarge);
        }
        let pending = end - start;
        let take = available.min(pending);
        self.buffer.as_mut()[filled..filled + take]
            .copy_from_slice(&self.plaintext[start..start + take]);
        self.filled = u16::try_from(filled + take).map_err(|_| Error::FrameTooLarge)?;
        self.plaintext_start = u16::try_from(start + take).map_err(|_| Error::FrameTooLarge)?;
        Ok(())
    }

    fn consume(&mut self, len: usize) {
        let filled = usize::from(self.filled);
        self.buffer.as_mut().copy_within(len..filled, 0);
        self.filled -= len as u16;
    }

    fn set_status(&mut self, stream: u32, status: u16) -> Result<(), Error<S::Error>> {
        self.statuses
            .set(stream, status)
            .map_err(|()| Error::ResponseTooLong)
    }

    fn take_status(&mut self, stream: u32) -> Option<u16> {
        self.statuses.take(stream)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MapEvent {
    Applied,
    KeepAlive,
}

/// Where one map document is accumulated. Lease mode keeps partial state in
/// a caller-owned working map so the public staging lease is only replaced
/// by complete snapshots; in-place mode accumulates directly in the caller's
/// single rotating lease, whose owner must not reseed it mid-document (see
/// [`MapStream::has_partial_response`]).
enum Target<'a, const PEERS: usize> {
    Lease {
        staging: &'a mut Netmap<PEERS>,
        working: &'a mut Netmap<PEERS>,
    },
    InPlace(&'a mut Netmap<PEERS>),
}

/// Persistent state for one long-lived streaming map response. The stream
/// itself retains no netmap: callers own both the staging lease and the
/// bounded working map a partial document is accumulated into, so each
/// consumer places that storage where its memory budget requires.
pub struct MapStream<const PEERS: usize> {
    stream: u32,
    frames: FrameReader,
    parser: Parser<PEERS>,
    document_started: bool,
    pending: heapless::Deque<MapEvent, 4>,
    status: Option<u16>,
    ended: bool,
}

impl<const PEERS: usize> Default for MapStream<PEERS> {
    fn default() -> Self {
        Self::new()
    }
}

impl<const PEERS: usize> MapStream<PEERS> {
    /// Construct inactive map-stream state. Firmware can place this in static
    /// storage and activate it with [`MapStream::start_in`] so the bounded JSON
    /// token buffer never becomes part of an async task's stack frame.
    pub const fn new() -> Self {
        Self {
            stream: 0,
            status: None,
            frames: FrameReader::new(),
            parser: Parser::new(),
            document_started: false,
            pending: heapless::Deque::new(),
            ended: false,
        }
    }

    /// Whether a response drained alongside another request already applied a
    /// newer map to its staging lease. Callers must not reseed that lease from
    /// the older active map before [`Self::next`] returns the queued event.
    /// A partial response is private and does not reserve the public lease.
    pub fn has_pending_update(&self) -> bool {
        self.pending.iter().any(|event| *event == MapEvent::Applied)
    }

    /// Whether a map document is partially accumulated. In-place callers
    /// (see [`Self::next_in_place`]) must not reseed or swap their lease
    /// while this is true: the parser's progress lives in that lease.
    pub fn has_partial_response(&self) -> bool {
        self.document_started
    }

    /// Wait until [`Self::next`] has a queued event or buffered input to
    /// process. This never consumes input, parses, or writes, so a caller may
    /// drop it to do other work on the connection and later call `next`
    /// without losing progress, provided the underlying stream's reads are
    /// cancellation-safe. `next` itself is not cancellation-safe: it writes
    /// HTTP/2 acknowledgements and flow-control updates while reading.
    pub async fn wait_input<S, B, C>(
        &self,
        http2: &mut Http2<S, B, C>,
    ) -> Result<(), Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        if !self.pending.is_empty() || self.ended {
            return Ok(());
        }
        http2.wait_readable().await
    }

    pub async fn start<S, B, C>(
        http2: &mut Http2<S, B, C>,
        authority: &str,
        request_body: &[u8],
    ) -> Result<Self, Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        let mut map = Self::new();
        Self::start_in(&mut map, http2, authority, request_body).await?;
        Ok(map)
    }

    /// Start a map request using caller-provided persistent storage.
    pub async fn start_in<S, B, C>(
        map: &mut Self,
        http2: &mut Http2<S, B, C>,
        authority: &str,
        request_body: &[u8],
    ) -> Result<(), Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        let stream = http2
            .send_request("POST", crate::control::map::PATH, authority, request_body)
            .await?;
        map.stream = stream;
        map.status = None;
        map.frames.finish_response();
        map.parser.reset();
        map.document_started = false;
        map.pending.clear();
        map.ended = false;
        Ok(())
    }

    /// Read the next map event, accumulating each document in `working`
    /// (seeded from `staging`) and publishing only complete snapshots to
    /// `staging`. Partial documents never mutate the staging lease, so
    /// callers may reseed it or drain interleaved responses at any point.
    pub async fn next<S, B, C>(
        &mut self,
        http2: &mut Http2<S, B, C>,
        staging: &mut Netmap<PEERS>,
        working: &mut Netmap<PEERS>,
    ) -> Result<MapEvent, Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        self.next_target(http2, Target::Lease { staging, working })
            .await
    }

    /// Read the next map event, accumulating each document directly in the
    /// caller's single rotating lease. The caller must have seeded the lease
    /// from the active map before the first document and must not reseed or
    /// swap it while [`Self::has_partial_response`] is true. `Applied` is
    /// only returned at a document boundary: input after the completed
    /// document stays unread, so the lease never carries a later partial one.
    pub async fn next_in_place<S, B, C>(
        &mut self,
        http2: &mut Http2<S, B, C>,
        lease: &mut Netmap<PEERS>,
    ) -> Result<MapEvent, Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        self.next_target(http2, Target::InPlace(lease)).await
    }

    async fn next_target<S, B, C>(
        &mut self,
        http2: &mut Http2<S, B, C>,
        mut target: Target<'_, PEERS>,
    ) -> Result<MapEvent, Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        loop {
            // A drained response may have completed one document and begun
            // the next in an in-place lease. The lease is not a snapshot
            // until that document finishes too; its Applied then coalesces.
            let lease_partial = self.document_started && matches!(target, Target::InPlace(_));
            if let Some(&event) = self.pending.front()
                && !(lease_partial && event == MapEvent::Applied)
            {
                self.pending.pop_front();
                return Ok(event);
            }
            if self.ended {
                return Err(Error::StreamEnded);
            }

            let input = http2.next_input().await?;
            if let Input::Data {
                stream,
                bytes,
                end_stream,
                frame_done,
            } = input
            {
                let len = bytes.len();
                let mut used = len;
                let mut stream_error = None;
                if stream == self.stream {
                    if self.status != Some(200) {
                        stream_error = Some(Error::UnexpectedStatus(self.status.unwrap_or(0)));
                    } else {
                        match self.queue_data(&mut target, bytes) {
                            Ok(queued) => used = queued,
                            Err(error) => stream_error = Some(error),
                        }
                    }
                    self.ended |= end_stream && frame_done && used == len;
                }
                http2.consume_data(used).await?;
                if let Some(error) = stream_error {
                    return Err(error);
                }
                continue;
            }
            let Input::Frame(frame_len) = input else {
                unreachable!();
            };
            let mut replies = [0; 256];
            let mut parsed_status = None;
            let (event, reply_len) = http2
                .connection
                .borrow_mut()
                .recv(
                    &http2.buffer.as_ref()[..frame_len],
                    |name, value| {
                        if name == ":status" {
                            parsed_status = Some(value.parse().unwrap_or(0));
                        }
                    },
                    &mut replies,
                )
                .map_err(Error::Http2)?;
            let result = self.consume_event(event, parsed_status, &mut target);
            http2.consume(frame_len);
            if reply_len != 0 {
                http2.send(&replies[..reply_len]).await?;
            }
            let (outcome, ended) = result?;
            if let Some(outcome) = outcome {
                return Ok(outcome);
            }
            if ended {
                return Err(Error::StreamEnded);
            }
        }
    }

    /// Drain another response while preserving and parsing interleaved map
    /// frames from this stream.
    pub async fn drain_response<S, B, C>(
        &mut self,
        http2: &mut Http2<S, B, C>,
        staging: &mut Netmap<PEERS>,
        working: &mut Netmap<PEERS>,
        wanted_stream: u32,
    ) -> Result<(), Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        self.drain_target(http2, Target::Lease { staging, working }, wanted_stream)
            .await
    }

    async fn drain_target<S, B, C>(
        &mut self,
        http2: &mut Http2<S, B, C>,
        mut target: Target<'_, PEERS>,
        wanted_stream: u32,
    ) -> Result<(), Error<S::Error>>
    where
        S: Read + Write,
        B: AsRef<[u8]> + AsMut<[u8]>,
        C: BorrowMut<Connection>,
    {
        loop {
            let input = http2.next_input().await?;
            if let Input::Data {
                stream,
                bytes,
                end_stream,
                frame_done,
            } = input
            {
                if stream != 0 && stream != wanted_stream && stream != self.stream {
                    return Err(Error::InterleavedStream(stream));
                }
                let len = bytes.len();
                let mut used = len;
                let mut stream_error = None;
                if stream == self.stream {
                    if self.status != Some(200) {
                        stream_error = Some(Error::UnexpectedStatus(self.status.unwrap_or(0)));
                    } else {
                        match self.queue_data(&mut target, bytes) {
                            Ok(queued) => used = queued,
                            Err(error) => stream_error = Some(error),
                        }
                    }
                    self.ended |= end_stream && frame_done && used == len;
                }
                let finished = stream == wanted_stream && end_stream && frame_done;
                http2.consume_data(used).await?;
                if let Some(error) = stream_error {
                    return Err(error);
                }
                if finished {
                    return Ok(());
                }
                continue;
            }
            let Input::Frame(frame_len) = input else {
                unreachable!();
            };
            let header = frame::FrameHeader::parse(&http2.buffer.as_ref()[..frame_len])
                .map_err(Error::Http2)?;
            if header.stream != 0 && header.stream != wanted_stream && header.stream != self.stream
            {
                return Err(Error::InterleavedStream(header.stream));
            }

            let mut replies = [0; 256];
            let mut parsed_status = None;
            let (event, reply_len) = http2
                .connection
                .borrow_mut()
                .recv(
                    &http2.buffer.as_ref()[..frame_len],
                    |name, value| {
                        if name == ":status" {
                            parsed_status = Some(value.parse().unwrap_or(0));
                        }
                    },
                    &mut replies,
                )
                .map_err(Error::Http2)?;
            let header_status = match &event {
                Event::Headers { stream, .. } => parsed_status.map(|status| (*stream, status)),
                _ => None,
            };
            let mut finished = false;
            let mut stream_error = None;
            match event {
                Event::Headers { stream, end_stream } if stream == wanted_stream => {
                    finished = end_stream;
                }
                Event::Data { .. } => unreachable!("DATA is streamed before frame dispatch"),
                Event::Headers { stream, end_stream } if stream == self.stream => {
                    if let Some(status) = parsed_status {
                        self.status = Some(status);
                    }
                    if self.status != Some(200) {
                        stream_error = Some(Error::UnexpectedStatus(self.status.unwrap_or(0)));
                    }
                    self.ended |= end_stream;
                }
                Event::Reset { stream, .. } if stream == wanted_stream || stream == self.stream => {
                    stream_error = Some(Error::Http2(micro_h2::Error::StreamReset));
                }
                Event::GoAway { .. } => {
                    stream_error = Some(Error::Http2(micro_h2::Error::GoAway));
                }
                _ => {}
            }
            if let Some((stream, status)) = header_status {
                http2.set_status(stream, status)?;
            }
            http2.consume(frame_len);
            if reply_len != 0 {
                http2.send(&replies[..reply_len]).await?;
            }
            if let Some(error) = stream_error {
                return Err(error);
            }
            if finished {
                return Ok(());
            }
        }
    }

    /// Feed `data` to the current document and return how much of it was
    /// used. An in-place target stops at a completed document: the bytes
    /// after it belong to the next document, which must not touch the lease
    /// before its owner has taken this one, so they stay unconsumed input.
    fn queue_data<E>(
        &mut self,
        target: &mut Target<'_, PEERS>,
        mut data: &[u8],
    ) -> Result<usize, Error<E>> {
        let total = data.len();
        while !data.is_empty() {
            let (outcome, consumed) = self.consume_data(target, data)?;
            data = &data[consumed..];
            if let Some(outcome) = outcome {
                // A single HTTP/2 DATA chunk may contain an arbitrary run of
                // zero-length keepalives or several complete deltas. Only the
                // latest applied map and the fact that liveness was observed
                // matter to the actor, so coalesce duplicate notifications
                // instead of turning valid compressed input into queue growth.
                if !self.pending.iter().any(|pending| *pending == outcome) {
                    self.pending
                        .push_back(outcome)
                        .map_err(|_| Error::ResponseTooLong)?;
                }
            }
            if consumed == 0 {
                return Err(Error::FrameTooLarge);
            }
            if outcome == Some(MapEvent::Applied) && matches!(target, Target::InPlace(_)) {
                break;
            }
        }
        Ok(total - data.len())
    }

    /// Push one body chunk into `target`; when the document completes,
    /// finish the parser there and report whether it set the self home.
    fn feed<E>(
        &mut self,
        target: &mut Netmap<PEERS>,
        bytes: &[u8],
    ) -> Result<Option<bool>, Error<E>> {
        self.parser.push(target, bytes).map_err(Error::Netmap)?;
        if self.frames.is_complete() {
            self.frames.finish_response();
            let parser = core::mem::take(&mut self.parser);
            let server_home = parser.sets_home_derp();
            self.document_started = false;
            parser.finish(target).map_err(Error::Netmap)?;
            return Ok(Some(server_home));
        }
        Ok(None)
    }

    fn consume_data<E>(
        &mut self,
        target: &mut Target<'_, PEERS>,
        data: &[u8],
    ) -> Result<(Option<MapEvent>, usize), Error<E>> {
        let mut consumed_total = 0;
        loop {
            let input = &data[consumed_total..];
            let (piece, consumed) = self.frames.next(input).map_err(Error::Netmap)?;
            consumed_total += consumed;
            match piece {
                crate::netmap::frame::Chunk::Body(bytes) => match target {
                    Target::Lease { staging, working } => {
                        if !self.document_started {
                            working.copy_from(staging);
                            self.document_started = true;
                        }
                        if let Some(server_home) = self.feed(working, bytes)? {
                            // The caller may have selected a home after this
                            // document began. Inherit that local preference
                            // unless this authenticated response explicitly
                            // replaces it.
                            if !server_home {
                                working.home_derp = staging.home_derp;
                            }
                            staging.copy_from(working);
                            return Ok((Some(MapEvent::Applied), consumed_total));
                        }
                    }
                    Target::InPlace(lease) => {
                        self.document_started = true;
                        if self.feed(lease, bytes)?.is_some() {
                            return Ok((Some(MapEvent::Applied), consumed_total));
                        }
                    }
                },
                crate::netmap::frame::Chunk::End => {
                    return Ok((Some(MapEvent::KeepAlive), consumed_total));
                }
                crate::netmap::frame::Chunk::Need if consumed == 0 => {
                    return Ok((None, consumed_total));
                }
                crate::netmap::frame::Chunk::Need => {}
            }
        }
    }

    /// Apply one decoded HTTP/2 event to this map response.
    ///
    /// Status is deliberately checked only on this stream's response HEADERS.
    /// A server is expected to send connection-level SETTINGS first; treating
    /// that frame as a response with status zero was the physical device's
    /// post-registration bootstrap failure.
    fn consume_event<E>(
        &mut self,
        event: Event<'_>,
        frame_status: Option<u16>,
        target: &mut Target<'_, PEERS>,
    ) -> Result<(Option<MapEvent>, bool), Error<E>> {
        match event {
            Event::Headers { stream, end_stream } if stream == self.stream => {
                if let Some(status) = frame_status {
                    self.status = Some(status);
                }
                if self.status != Some(200) {
                    return Err(Error::UnexpectedStatus(self.status.unwrap_or(0)));
                }
                self.ended |= end_stream;
            }
            Event::Data {
                stream,
                data,
                end_stream,
            } if stream == self.stream => {
                if self.status != Some(200) {
                    return Err(Error::UnexpectedStatus(self.status.unwrap_or(0)));
                }
                // Whole-frame DATA dispatch is never chosen by the streamed
                // readers above, so there is no input to leave a suffix in.
                let mut rest = data;
                while !rest.is_empty() {
                    let used = self.queue_data(target, rest)?;
                    rest = &rest[used..];
                }
                self.ended |= end_stream;
                return Ok((self.pending.pop_front(), self.ended));
            }
            Event::Reset { stream, .. } if stream == self.stream => {
                return Err(Error::Http2(micro_h2::Error::StreamReset));
            }
            Event::GoAway { .. } => return Err(Error::Http2(micro_h2::Error::GoAway)),
            _ => {}
        }
        Ok((None, false))
    }
}

/// Send a non-streaming endpoint update without disturbing the map stream.
pub async fn send_endpoint_update<S, B, C, const PEERS: usize>(
    http2: &mut Http2<S, B, C>,
    map: &mut MapStream<PEERS>,
    staging: &mut Netmap<PEERS>,
    working: &mut Netmap<PEERS>,
    authority: &str,
    request_body: &[u8],
) -> Result<(), Error<S::Error>>
where
    S: Read + Write,
    B: AsRef<[u8]> + AsMut<[u8]>,
    C: BorrowMut<Connection>,
{
    send_endpoint_update_target(
        http2,
        map,
        Target::Lease { staging, working },
        authority,
        request_body,
    )
    .await
}

/// In-place form of [`send_endpoint_update`] for a single rotating lease;
/// the same mid-document discipline as [`MapStream::next_in_place`] applies.
pub async fn send_endpoint_update_in_place<S, B, C, const PEERS: usize>(
    http2: &mut Http2<S, B, C>,
    map: &mut MapStream<PEERS>,
    lease: &mut Netmap<PEERS>,
    authority: &str,
    request_body: &[u8],
) -> Result<(), Error<S::Error>>
where
    S: Read + Write,
    B: AsRef<[u8]> + AsMut<[u8]>,
    C: BorrowMut<Connection>,
{
    send_endpoint_update_target(http2, map, Target::InPlace(lease), authority, request_body).await
}

async fn send_endpoint_update_target<S, B, C, const PEERS: usize>(
    http2: &mut Http2<S, B, C>,
    map: &mut MapStream<PEERS>,
    target: Target<'_, PEERS>,
    authority: &str,
    request_body: &[u8],
) -> Result<(), Error<S::Error>>
where
    S: Read + Write,
    B: AsRef<[u8]> + AsMut<[u8]>,
    C: BorrowMut<Connection>,
{
    let stream = http2
        .send_request("POST", crate::control::map::PATH, authority, request_body)
        .await?;
    map.drain_target(http2, target, stream).await?;
    let status = http2.take_status(stream).unwrap_or(0);
    if status != 200 {
        return Err(Error::UnexpectedStatus(status));
    }
    Ok(())
}

fn json_string<'a>(body: &'a [u8], field: &str) -> Option<&'a str> {
    let text = core::str::from_utf8(body).ok()?;
    let mut pattern = heapless::String::<64>::new();
    pattern.push('"').ok()?;
    pattern.push_str(field).ok()?;
    pattern.push('"').ok()?;
    let rest = &text[text.find(pattern.as_str())? + pattern.len()..];
    let rest = rest.trim_start().strip_prefix(':')?.trim_start();
    let rest = rest.strip_prefix('"')?;
    Some(&rest[..rest.find('"')?])
}

const _: () = assert!(CAPABILITY_VERSION == 131);

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

    struct PausingReader {
        input: &'static [u8],
        position: usize,
        pause_at: usize,
        paused: bool,
    }

    impl embedded_io_async::ErrorType for PausingReader {
        type Error = core::convert::Infallible;
    }

    impl Read for PausingReader {
        async fn read(&mut self, output: &mut [u8]) -> Result<usize, Self::Error> {
            let limit = if self.paused {
                self.pause_at
            } else {
                self.input.len()
            };
            if self.position == limit {
                return core::future::pending().await;
            }
            let amount = output.len().min(limit - self.position);
            output[..amount].copy_from_slice(&self.input[self.position..self.position + amount]);
            self.position += amount;
            Ok(amount)
        }
    }

    #[test]
    fn exact_read_progress_survives_future_cancellation() {
        use core::future::Future as _;
        use core::task::{Context, Poll, Waker};

        let mut stream = PausingReader {
            input: b"record",
            position: 0,
            pause_at: 2,
            paused: true,
        };
        let mut output = [0; 6];
        let mut filled = 0;
        let waker = Waker::noop();
        let mut context = Context::from_waker(waker);

        {
            let mut read =
                core::pin::pin!(read_exact_progress(&mut stream, &mut output, &mut filled));
            assert!(matches!(read.as_mut().poll(&mut context), Poll::Pending));
        }
        assert_eq!(filled, 2);

        stream.paused = false;
        {
            let mut read =
                core::pin::pin!(read_exact_progress(&mut stream, &mut output, &mut filled));
            assert!(matches!(
                read.as_mut().poll(&mut context),
                Poll::Ready(Ok(()))
            ));
        }
        assert_eq!(&output, b"record");
    }

    #[test]
    fn compact_status_storage_preserves_eight_active_streams() {
        let mut statuses = Statuses::new();
        for stream in 1..=H2_STATUS_SLOTS as u32 {
            statuses.set(stream, 200 + stream as u16).unwrap();
        }
        assert!(statuses.set(9, 209).is_err());
        assert_eq!(statuses.get(4), Some(204));
        assert_eq!(statuses.take(4), Some(204));
        statuses.set(9, 209).unwrap();
        assert_eq!(statuses.get(9), Some(209));
    }

    #[test]
    fn map_boundaries_stop_before_the_next_response_prefix() {
        let mut stream = MapStream::<1> {
            stream: 1,
            frames: FrameReader::new(),
            parser: Parser::new(),
            document_started: false,
            pending: heapless::Deque::new(),
            status: Some(200),
            ended: false,
        };
        let mut staging = Netmap::new();
        let mut working = Netmap::new();
        let bytes = [0u8; 8];

        let (first, consumed) = stream
            .consume_data::<()>(
                &mut Target::Lease {
                    staging: &mut staging,
                    working: &mut working,
                },
                &bytes,
            )
            .unwrap();
        assert_eq!(first, Some(MapEvent::KeepAlive));
        assert_eq!(consumed, 4);
        let (second, consumed) = stream
            .consume_data::<()>(
                &mut Target::Lease {
                    staging: &mut staging,
                    working: &mut working,
                },
                &bytes[consumed..],
            )
            .unwrap();
        assert_eq!(second, Some(MapEvent::KeepAlive));
        assert_eq!(consumed, 4);
    }

    #[derive(serde::Deserialize)]
    struct PostRegisterFixture {
        map_stream: u32,
        frames: std::vec::Vec<std::string::String>,
    }

    fn decode_hex(text: &str) -> std::vec::Vec<u8> {
        assert!(text.len().is_multiple_of(2));
        text.as_bytes()
            .chunks_exact(2)
            .map(|digits| {
                let digits = core::str::from_utf8(digits).unwrap();
                u8::from_str_radix(digits, 16).unwrap()
            })
            .collect()
    }

    #[test]
    fn explicit_http_body_boundary_does_not_require_connection_close() {
        let response = b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nkey";
        assert_eq!(response_length(response).unwrap(), Some(response.len()));
        assert_eq!(
            response_length(&response[..response.len() - 1]).unwrap(),
            Some(response.len())
        );
    }

    #[test]
    fn close_delimited_and_malformed_lengths_are_handled_safely() {
        assert_eq!(
            response_length(b"HTTP/1.1 200 OK\r\n\r\nkey").unwrap(),
            None
        );
        assert!(response_length(b"HTTP/1.1 200 OK\r\nContent-Length: no\r\n\r\n").is_err());
        assert!(
            response_length(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\nContent-Length: 1\r\n\r\nx")
                .is_err()
        );
    }

    #[test]
    fn settings_before_map_headers_are_not_status_zero() {
        let fixture: PostRegisterFixture = serde_json::from_str(include_str!(
            "../../../../tests/vectors/post_register_map_h2.json"
        ))
        .unwrap();
        let mut connection = Connection::new();
        // The capture starts after registration and the map request. Open the
        // same streams before replaying their responses.
        let mut request = [0; 256];
        for path in ["/machine/register", "/machine/map"] {
            connection
                .request(
                    "POST",
                    path,
                    "control.example",
                    "https",
                    &[],
                    &[],
                    &mut request,
                )
                .unwrap();
        }
        let mut map = MapStream::<1> {
            stream: fixture.map_stream,
            status: None,
            frames: FrameReader::new(),
            parser: Parser::new(),
            document_started: false,
            pending: heapless::Deque::new(),
            ended: false,
        };
        let mut netmap = Netmap::new();
        let mut working = Netmap::new();
        let mut applied = false;

        for encoded in fixture.frames {
            let bytes = decode_hex(&encoded);
            let mut status = None;
            let mut reply = [0; 256];
            let (event, _) = connection
                .recv(
                    &bytes,
                    |name, value| {
                        if name == ":status" {
                            status = value.parse().ok();
                        }
                    },
                    &mut reply,
                )
                .unwrap();
            let (outcome, ended) = map
                .consume_event::<core::convert::Infallible>(
                    event,
                    status,
                    &mut Target::Lease {
                        staging: &mut netmap,
                        working: &mut working,
                    },
                )
                .unwrap();
            applied |= outcome == Some(MapEvent::Applied);
            assert!(!ended);
        }

        assert_eq!(map.status, Some(200));
        assert!(applied);
        assert_eq!(netmap.responses, 1);
    }

    /// An in-memory stream: reads drain `inbound`, writes append to `written`.
    struct Scripted {
        inbound: std::vec::Vec<u8>,
        read: usize,
        written: std::vec::Vec<u8>,
        available: Option<usize>,
        pause_at_end: bool,
    }

    impl Scripted {
        fn new(inbound: std::vec::Vec<u8>) -> Self {
            Self {
                inbound,
                read: 0,
                written: std::vec::Vec::new(),
                available: None,
                pause_at_end: false,
            }
        }
    }

    impl embedded_io_async::ErrorType for Scripted {
        type Error = core::convert::Infallible;
    }

    impl Read for Scripted {
        async fn read(&mut self, out: &mut [u8]) -> Result<usize, Self::Error> {
            let available = self.available.unwrap_or(self.inbound.len());
            if self.read == available && self.pause_at_end {
                return core::future::pending().await;
            }
            let take = out.len().min(available - self.read);
            out[..take].copy_from_slice(&self.inbound[self.read..self.read + take]);
            self.read += take;
            Ok(take)
        }
    }

    impl Write for Scripted {
        async fn write(&mut self, bytes: &[u8]) -> Result<usize, Self::Error> {
            self.written.extend_from_slice(bytes);
            Ok(bytes.len())
        }

        async fn flush(&mut self) -> Result<(), Self::Error> {
            Ok(())
        }
    }

    fn block_on<F: core::future::Future>(future: F) -> F::Output {
        let mut future = core::pin::pin!(future);
        let mut context = core::task::Context::from_waker(core::task::Waker::noop());
        match future.as_mut().poll(&mut context) {
            core::task::Poll::Ready(output) => output,
            core::task::Poll::Pending => panic!("scripted streams never block"),
        }
    }

    // Native policy storage is deliberately large even for a two-peer map.
    // Keep fixture values behind separate heap boundaries so debug builds do
    // not reserve several complete map temporaries in one test entry frame.
    #[inline(never)]
    fn lease_netmap<const PEERS: usize>() -> std::boxed::Box<Netmap<PEERS>> {
        std::boxed::Box::new(Netmap::new())
    }

    #[inline(never)]
    fn lease_empty_map<const PEERS: usize>() -> std::boxed::Box<MapStream<PEERS>> {
        std::boxed::Box::new(MapStream::new())
    }

    fn lease_stream<const PEERS: usize>(
        http: &mut Http2<Scripted>,
    ) -> std::boxed::Box<MapStream<PEERS>> {
        let mut map = lease_empty_map();
        block_on(MapStream::start_in(
            &mut map,
            http,
            "control.example",
            b"{}",
        ))
        .unwrap();
        map
    }

    fn lease_peer_response(id: u64, endpoint: &str) -> std::string::String {
        std::format!(
            "{{\"PeersChanged\":[{{\"ID\":{id},\"Key\":\"nodekey:{}\",\"Endpoints\":[\"{endpoint}\"]}}],\"KeepAlive\":false}}",
            std::format!("{:02x}", id as u8).repeat(32)
        )
    }

    fn lease_map_bytes(json: &str) -> std::vec::Vec<u8> {
        let mut bytes = std::vec::Vec::from((json.len() as u32).to_le_bytes());
        bytes.extend_from_slice(json.as_bytes());
        bytes
    }

    fn lease_headers(stream: u32, end_stream: bool) -> std::vec::Vec<u8> {
        let mut headers = [0; 32];
        let len =
            micro_h2::hpack::encode::encode_header(":status", "200", &mut headers, 0).unwrap();
        let mut bytes = [0; 64];
        let len = frame::write_frame(
            micro_h2::FrameType::Headers,
            frame::flags::END_HEADERS
                | if end_stream {
                    frame::flags::END_STREAM
                } else {
                    0
                },
            stream,
            &headers[..len],
            &mut bytes,
        )
        .unwrap();
        std::vec::Vec::from(&bytes[..len])
    }

    fn lease_data(bytes: &[u8]) -> std::vec::Vec<u8> {
        let mut output = std::vec![0; bytes.len() + 9];
        let len = frame::write_frame(micro_h2::FrameType::Data, 0, 1, bytes, &mut output).unwrap();
        output.truncate(len);
        output
    }

    fn lease_http(frames: &[std::vec::Vec<u8>]) -> Http2<Scripted> {
        // Authenticated, stateful Noise records carry genuine HTTP/2 frames.
        // The endpoint response is stream3, after the long-lived map stream1.
        let mut server = Session::new([2; 32], [1; 32]);
        let mut inbound = std::vec::Vec::new();
        for plaintext in frames {
            for chunk in plaintext.chunks(noise::MAX_PLAINTEXT) {
                let mut record = [0; noise::MAX_MESSAGE];
                let len = server.seal(chunk, &mut record).unwrap();
                inbound.extend_from_slice(&record[..len]);
            }
        }
        block_on(Http2::start(
            Scripted::new(inbound),
            Session::new([1; 32], [2; 32]),
            &[],
        ))
        .unwrap()
    }

    fn lease_available_frames(
        http: &mut Http2<Scripted>,
        frames: &[std::vec::Vec<u8>],
        count: usize,
    ) {
        http.stream.available = Some(
            frames[..count]
                .iter()
                .map(|bytes| {
                    bytes.len()
                        + bytes.chunks(noise::MAX_PLAINTEXT).count()
                            * (noise::HEADER_LEN + noise::TAG_LEN)
                })
                .sum(),
        );
        http.stream.pause_at_end = true;
    }

    #[test]
    fn map_lease_a_complete_map_is_prompt_while_the_next_length_prefix_is_paused() {
        for prefix in 1..=4 {
            let a = lease_peer_response(7, "198.51.100.7:4007");
            let b = lease_map_bytes(&lease_peer_response(8, "198.51.100.8:4008"));
            let mut combined = lease_map_bytes(&a);
            combined.extend_from_slice(&b[..prefix]);
            let mut http = lease_http(&[
                lease_headers(1, false),
                lease_data(&combined),
                lease_headers(3, true),
            ]);
            http.stream.pause_at_end = true;
            let mut map = lease_stream::<2>(&mut http);
            let mut staging = lease_netmap::<2>();
            let mut working = lease_netmap::<2>();
            block_on(send_endpoint_update(
                &mut http,
                &mut map,
                &mut staging,
                &mut working,
                "control.example",
                b"{}",
            ))
            .unwrap();
            // There is no suffix yet. block_on requires this poll to finish,
            // so waiting for the unrelated next document cannot pass.
            assert_eq!(
                block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
                MapEvent::Applied
            );
            assert_eq!(staging.responses, 1);
            assert!(staging.peers.get(7).is_some());
            assert!(staging.peers.get(8).is_none());
        }
    }

    #[test]
    fn map_lease_continuous_partial_documents_expose_each_ready_complete_snapshot() {
        let responses: std::vec::Vec<_> = (1..=4)
            .map(|port| lease_peer_response(7, &std::format!("198.51.100.7:{}", 4000 + port)))
            .collect();
        let bodies: std::vec::Vec<_> = responses.iter().map(|r| lease_map_bytes(r)).collect();
        let splits: std::vec::Vec<_> = responses
            .iter()
            .map(|r| 4 + r.find("\"KeepAlive\"").unwrap())
            .collect();
        let mut first = bodies[0].clone();
        first.extend_from_slice(&bodies[1][..splits[1]]);
        let mut second = bodies[1][splits[1]..].to_vec();
        second.extend_from_slice(&bodies[2][..splits[2]]);
        let mut third = bodies[2][splits[2]..].to_vec();
        third.extend_from_slice(&bodies[3][..splits[3]]);
        let mut fourth = bodies[3][splits[3]..].to_vec();
        fourth.extend_from_slice(&[0; 4]);
        let frames = [
            lease_headers(1, false),
            lease_data(&first),
            lease_headers(3, true),
            lease_data(&second),
            lease_data(&third),
            lease_data(&fourth),
        ];
        let mut http = lease_http(&frames);
        lease_available_frames(&mut http, &frames, 3);
        let mut map = lease_stream::<2>(&mut http);
        let mut active = lease_netmap::<2>();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        block_on(send_endpoint_update(
            &mut http,
            &mut map,
            &mut staging,
            &mut working,
            "control.example",
            b"{}",
        ))
        .unwrap();
        for completed in 1..=4 {
            if !map.has_pending_update() {
                staging.copy_from(&active);
            }
            lease_available_frames(&mut http, &frames, completed + 2);
            assert_eq!(
                block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
                MapEvent::Applied,
                "a complete map must remain prompt even when every chunk ends with the next partial document"
            );
            core::mem::swap(&mut active, &mut staging);
            assert_eq!(active.responses, completed);
            assert_eq!(
                active.peers.get(7).unwrap().endpoints.as_slice(),
                &[std::format!("198.51.100.7:{}", 4000 + completed)
                    .parse()
                    .unwrap()]
            );
        }
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::KeepAlive
        );
    }

    #[inline(never)]
    fn lease_new_map256() -> std::boxed::Box<MapStream<256>> {
        lease_empty_map()
    }

    #[inline(never)]
    fn lease_owned_start_future_size() -> usize {
        let mut http = lease_http(&[]);
        let future = MapStream::<256>::start(&mut http, "control.example", b"{}");
        core::mem::size_of_val(&future)
    }

    #[test]
    fn map_lease_storage_layout_budget() {
        use core::mem::{size_of, size_of_val};
        for (peers, map, stream) in [
            (2, size_of::<Netmap<2>>(), size_of::<MapStream<2>>()),
            (16, size_of::<Netmap<16>>(), size_of::<MapStream<16>>()),
            (32, size_of::<Netmap<32>>(), size_of::<MapStream<32>>()),
            (128, size_of::<Netmap<128>>(), size_of::<MapStream<128>>()),
            (256, size_of::<Netmap<256>>(), size_of::<MapStream<256>>()),
        ] {
            std::println!(
                "map_lease_layout peers={peers} netmap_bytes={map} stream_bytes={stream} hosted_policy={} peer_names={} ssh_policy={}",
                cfg!(feature = "hosted-policy"),
                cfg!(feature = "peer-names"),
                cfg!(feature = "ssh-policy")
            );
        }
        std::println!(
            "map_lease_layout peers=256 start_future_bytes={}",
            lease_owned_start_future_size()
        );
        let mut http = lease_http(&[]);
        let mut place = std::hint::black_box(lease_new_map256());
        let reset = MapStream::start_in(&mut place, &mut http, "control.example", b"{}");
        std::println!(
            "map_lease_layout peers=256 start_in_future_bytes={}",
            size_of_val(&reset)
        );
        block_on(reset).unwrap();
    }

    #[test]
    fn map_lease_local_home_changes_do_not_override_an_explicit_response_home() {
        for (self_record, expected) in [
            ("", 42),
            ("\"Node\":{\"HomeDERP\":17},", 17),
            ("\"Node\":{\"HomeDERP\":0},", 0),
            ("\"Node\":{\"DERPRegion\":23},", 23),
            ("\"Node\":{\"DERP\":\"127.3.3.40:29\"},", 29),
            ("\"Node\":{\"DERP\":\"127.3.3.29:1\"},", 42),
            ("\"Node\":{\"HomeDERP\":17,\"DERP\":\"127.3.3.40:29\"},", 17),
            ("\"Node\":{\"DERP\":\"127.3.3.40:29\",\"HomeDERP\":17},", 17),
        ] {
            let peers = lease_peer_response(7, "198.51.100.7:4007");
            let response = std::format!("{{{self_record}{}", &peers[1..]);
            let split = 4 + response.find("\"KeepAlive\"").unwrap();
            let body = lease_map_bytes(&response);
            let mut http = lease_http(&[
                lease_headers(1, false),
                lease_data(&body[..split]),
                lease_headers(3, true),
                lease_data(&body[split..]),
            ]);
            let mut map = lease_stream::<2>(&mut http);
            let mut active = lease_netmap::<2>();
            active.home_derp = 17;
            let mut staging = lease_netmap::<2>();
            let mut working = lease_netmap::<2>();
            staging.copy_from(&active);
            block_on(send_endpoint_update(
                &mut http,
                &mut map,
                &mut staging,
                &mut working,
                "control.example",
                b"{}",
            ))
            .unwrap();
            active.home_derp = 42; // Same active-map mutation as prefer_derp.
            if !map.has_pending_update() {
                staging.copy_from(&active);
            }
            assert_eq!(
                block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
                MapEvent::Applied
            );
            assert_eq!(
                staging.home_derp, expected,
                "response-scoped home precedence: {self_record}"
            );
        }
    }

    #[test]
    fn map_lease_a_cancelled_next_retains_private_body_and_noise_read_progress() {
        use core::future::Future;
        use core::task::{Context, Poll, Waker};
        let response = lease_peer_response(7, "198.51.100.7:4007");
        let split = 4 + response.find("\"KeepAlive\"").unwrap();
        let body = lease_map_bytes(&response);
        let frames = [
            lease_headers(1, false),
            lease_data(&body[..split]),
            lease_data(&body[split..]),
        ];
        let mut http = lease_http(&frames);
        lease_available_frames(&mut http, &frames, 2);
        // Permit one byte of the next encrypted record header, then pause.
        // Both the map parser and HTTP/Noise read now own incomplete progress.
        http.stream.available = http.stream.available.map(|limit| limit + 1);
        let mut map = lease_stream::<2>(&mut http);
        let mut active = lease_netmap::<2>();
        active
            .peers
            .upsert(crate::netmap::Peer::new(
                1,
                crate::keys::NodePublic::from_bytes([1; 32]),
            ))
            .unwrap();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        staging.copy_from(&active);
        {
            let mut next = core::pin::pin!(map.next(&mut http, &mut staging, &mut working));
            assert!(matches!(
                next.as_mut().poll(&mut Context::from_waker(Waker::noop())),
                Poll::Pending
            ));
        }
        assert!(
            !map.has_pending_update(),
            "private partial parsing must not reserve public staging"
        );
        assert!(map.has_partial_response());
        staging.copy_from(&active);
        lease_available_frames(&mut http, &frames, 3);
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        assert!(staging.peers.get(1).is_some());
        assert!(
            staging.peers.get(7).is_some(),
            "dropping next and reseeding public staging must not lose private peer tokens"
        );
    }

    #[test]
    fn map_lease_wait_input_is_read_only_and_cancellation_safe() {
        use core::future::Future;
        use core::task::{Context, Poll, Waker};
        let mut context = Context::from_waker(Waker::noop());
        let frames = [lease_headers(1, false), lease_data(&[0; 4])];
        let mut http = lease_http(&frames);
        lease_available_frames(&mut http, &frames, 1);
        let mut map = lease_stream::<2>(&mut http);
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        // Consume the response headers; the stream is then quiet.
        {
            let mut next = core::pin::pin!(map.next(&mut http, &mut staging, &mut working));
            assert!(next.as_mut().poll(&mut context).is_pending());
        }
        let written = http.stream.written.len();
        {
            let mut wait = core::pin::pin!(map.wait_input(&mut http));
            assert!(wait.as_mut().poll(&mut context).is_pending());
        }
        // Expose one byte of the keepalive's encrypted record header, so the
        // dropped wait owns partial Noise record progress.
        http.stream.available = http.stream.available.map(|limit| limit + 1);
        {
            let mut wait = core::pin::pin!(map.wait_input(&mut http));
            assert!(wait.as_mut().poll(&mut context).is_pending());
        }
        lease_available_frames(&mut http, &frames, 2);
        {
            let mut wait = core::pin::pin!(map.wait_input(&mut http));
            assert!(matches!(
                wait.as_mut().poll(&mut context),
                Poll::Ready(Ok(()))
            ));
        }
        // Buffered input stays ready until `next` consumes it.
        assert!(matches!(
            core::pin::pin!(map.wait_input(&mut http))
                .as_mut()
                .poll(&mut context),
            Poll::Ready(Ok(()))
        ));
        assert_eq!(
            http.stream.written.len(),
            written,
            "waiting for map input must never write"
        );
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::KeepAlive
        );
    }

    #[test]
    fn map_lease_multiple_complete_responses_coalesce_without_a_stale_swap() {
        let mut combined = lease_map_bytes(&lease_peer_response(7, "198.51.100.7:4007"));
        combined.extend_from_slice(&lease_map_bytes(&lease_peer_response(
            7,
            "203.0.113.7:5007",
        )));
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&combined),
            lease_headers(3, true),
            lease_data(&[0; 4]),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let mut active = lease_netmap::<2>();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        block_on(send_endpoint_update(
            &mut http,
            &mut map,
            &mut staging,
            &mut working,
            "control.example",
            b"{}",
        ))
        .unwrap();
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        core::mem::swap(&mut active, &mut staging);
        assert_eq!(active.responses, 2);
        assert_eq!(
            active.peers.get(7).unwrap().endpoints.as_slice(),
            &["203.0.113.7:5007".parse().unwrap()]
        );
        staging.copy_from(&active);
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::KeepAlive,
            "the coalesced notification must not be replayed after the caller swaps the completed lease"
        );
        assert_eq!(active.responses, 2);
    }

    #[test]
    fn map_lease_partial_response_survives_an_endpoint_ack_and_caller_reseed() {
        let response = lease_peer_response(7, "198.51.100.7:4007");
        let split = response.find("\"KeepAlive\"").unwrap();
        let body = lease_map_bytes(&response);
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&body[..4 + split]),
            lease_headers(3, true),
            lease_data(&body[4 + split..]),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let active = lease_netmap::<2>();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        staging.copy_from(&active);
        block_on(send_endpoint_update(
            &mut http,
            &mut map,
            &mut staging,
            &mut working,
            "control.example",
            b"{}",
        ))
        .unwrap();
        // The parser consumed an early peer, but partial state is now private.
        assert!(!map.frames.is_complete(), "the suffix must still be unread");
        assert!(map.has_partial_response());
        // This is the production caller's lease guard, not a reset of the
        // parser. An endpoint ACK must not make an in-progress lease reusable.
        if !map.has_pending_update() {
            staging.copy_from(&active);
        }
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        assert!(
            staging.peers.get(7).is_some(),
            "endpoint ACK/reseed erased an already parsed peer before the map suffix completed"
        );
    }

    #[test]
    fn map_lease_complete_then_partial_response_never_publishes_the_partial_map() {
        let response_a = lease_peer_response(7, "198.51.100.7:4007");
        let response_b = lease_peer_response(7, "203.0.113.7:5007");
        let split = response_b.find("\"KeepAlive\"").unwrap();
        let body_b = lease_map_bytes(&response_b);
        let mut combined = lease_map_bytes(&response_a);
        combined.extend_from_slice(&body_b[..4 + split]);
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&combined),
            lease_headers(3, true),
            lease_data(&body_b[4 + split..]),
            lease_data(&[0; 4]),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let mut active = lease_netmap::<2>();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        staging.copy_from(&active);
        block_on(send_endpoint_update(
            &mut http,
            &mut map,
            &mut staging,
            &mut working,
            "control.example",
            b"{}",
        ))
        .unwrap();
        assert!(map.has_pending_update());
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        core::mem::swap(&mut active, &mut staging);
        assert_eq!(
            active.peers.get(7).unwrap().endpoints.as_slice(),
            &["198.51.100.7:4007".parse().unwrap()],
            "Applied(A) exposed peer tokens from the incomplete response B"
        );
        assert_eq!(active.responses, 1);
        if !map.has_pending_update() {
            staging.copy_from(&active);
        }
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        core::mem::swap(&mut active, &mut staging);
        assert_eq!(active.responses, 2);
        assert_eq!(
            active.peers.get(7).unwrap().endpoints.as_slice(),
            &["203.0.113.7:5007".parse().unwrap()]
        );
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::KeepAlive,
            "a duplicate Applied event would swap a stale staging lease back into active state"
        );
        assert_eq!(
            active.peers.get(7).unwrap().endpoints.as_slice(),
            &["203.0.113.7:5007".parse().unwrap()]
        );
    }

    #[test]
    fn map_lease_complete_response_remains_guarded_across_an_endpoint_ack() {
        let response = lease_peer_response(7, "198.51.100.7:4007");
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&lease_map_bytes(&response)),
            lease_headers(3, true),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let active = lease_netmap::<2>();
        let mut staging = lease_netmap::<2>();
        let mut working = lease_netmap::<2>();
        block_on(send_endpoint_update(
            &mut http,
            &mut map,
            &mut staging,
            &mut working,
            "control.example",
            b"{}",
        ))
        .unwrap();
        if !map.has_pending_update() {
            staging.copy_from(&active);
        }
        assert_eq!(
            block_on(map.next(&mut http, &mut staging, &mut working)).unwrap(),
            MapEvent::Applied
        );
        assert!(
            staging.peers.get(7).is_some(),
            "complete-response control must retain its peer"
        );
    }

    #[test]
    fn streamed_data_validates_content_length_without_counting_padding() {
        // Every attempt reuses the same protocol state, including after an
        // invalid response. Each new session must start again at stream 1.
        let mut buffer = [0; H2_BUFFER];
        let mut connection = Connection::new();
        for padding in [None, Some(0), Some(7), Some(255)] {
            for content_length in [4, 5, 6] {
                let mut headers = [0; 64];
                let len = micro_h2::hpack::encode::encode_header(":status", "200", &mut headers, 0)
                    .unwrap();
                let len = micro_h2::hpack::encode::encode_header(
                    "content-length",
                    &std::format!("{content_length}"),
                    &mut headers,
                    len,
                )
                .unwrap();
                let mut response = [0; 512];
                let mut len = frame::write_frame(
                    micro_h2::FrameType::Headers,
                    frame::flags::END_HEADERS,
                    1,
                    &headers[..len],
                    &mut response,
                )
                .unwrap();
                let mut payload = std::vec::Vec::new();
                if let Some(padding) = padding {
                    payload.push(padding);
                }
                payload.extend_from_slice(b"hello");
                if let Some(padding) = padding {
                    payload.resize(payload.len() + usize::from(padding), 0);
                }
                len += frame::write_frame(
                    micro_h2::FrameType::Data,
                    frame::flags::END_STREAM
                        | if padding.is_some() {
                            frame::flags::PADDED
                        } else {
                            0
                        },
                    1,
                    &payload,
                    &mut response[len..],
                )
                .unwrap();
                let mut server = Session::new([2; 32], [1; 32]);
                let mut inbound = std::vec::Vec::new();
                // Split both DATA and its padding across authenticated records.
                for chunk in response[..len].chunks(3) {
                    let mut record = [0; 64];
                    let len = server.seal(chunk, &mut record).unwrap();
                    inbound.extend_from_slice(&record[..len]);
                }
                let noise = NoiseStream {
                    stream: Scripted::new(inbound),
                    session: Session::new([1; 32], [2; 32]),
                    pushback: heapless::Vec::new(),
                };
                let mut http2 =
                    block_on(noise.into_https2_with_storage(&mut buffer, &mut connection)).unwrap();
                let mut body = [0; 5];
                let result = block_on(http2.request(
                    "POST",
                    "/machine/register",
                    "control.example",
                    &[],
                    &mut body,
                ));
                if content_length == 5 {
                    assert_eq!(result.unwrap(), (200, 5), "padding: {padding:?}");
                    assert_eq!(&body, b"hello");
                } else {
                    assert!(matches!(
                        result,
                        Err(Error::Http2(micro_h2::Error::Protocol))
                    ));
                }
            }
        }
    }

    struct Counter(u8);

    impl crate::keys::Rng for Counter {
        fn fill_bytes(&mut self, out: &mut [u8]) {
            for byte in out {
                *byte = self.0;
                self.0 = self.0.wrapping_add(1);
            }
        }
    }

    /// The server's side after the Noise response: an early payload and the
    /// first nine bytes of its HTTP/2 preface, split across two records.
    fn server_records(server: &mut Session) -> (std::vec::Vec<u8>, [u8; 9]) {
        let challenge = br#"{"nodeKeyChallenge":"chalpub:00"}"#;
        let settings = [0, 0, 0, 4, 0, 0, 0, 0, 0];
        let mut first = std::vec::Vec::from(noise::EARLY_MAGIC);
        first.extend_from_slice(&(challenge.len() as u32).to_be_bytes());
        first.extend_from_slice(&challenge[..10]);
        let mut second = std::vec::Vec::from(&challenge[10..]);
        second.extend_from_slice(&settings);
        let mut wire = std::vec::Vec::new();
        for plaintext in [first, second] {
            let mut record = [0; noise::MAX_MESSAGE];
            let len = server.seal(&plaintext, &mut record).unwrap();
            wire.extend_from_slice(&record[..len]);
        }
        (wire, settings)
    }

    #[test]
    fn a_split_handshake_yields_the_same_session_as_the_http_upgrade() {
        let identity = Identity::generate(&mut Counter(1));
        let server_private = StaticSecret::from([0x42; 32]);
        let server_key =
            MachinePublic::from_bytes(x25519_dalek::PublicKey::from(&server_private).to_bytes());
        let server_ephemeral = StaticSecret::from([0x43; 32]);

        // WebSocket shape: the initiation leaves out of band, and the stream
        // starts directly with the Noise response.
        let (initiation, pending) = initiate(&identity, &server_key, &mut Counter(9)).unwrap();
        let (response, mut server_split) = crate::noise::ik::respond_for_test(
            &server_private,
            &server_ephemeral,
            initiation.as_bytes(),
        );
        let mut inbound = std::vec::Vec::from(response);
        let (records, settings) = server_records(&mut server_split);
        inbound.extend_from_slice(&records);
        let split = block_on(complete(Scripted::new(inbound), pending)).unwrap();
        assert!(split.stream.written.is_empty());
        let mut split = block_on(split.into_http2()).unwrap();

        // HTTP/1.1 shape, from the same random stream.
        let (response, mut server_upgrade) = crate::noise::ik::respond_for_test(
            &server_private,
            &server_ephemeral,
            initiation.as_bytes(),
        );
        let mut inbound = std::vec::Vec::from(
            &b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: tailscale-control-protocol\r\nConnection: upgrade\r\n\r\n"[..],
        );
        inbound.extend_from_slice(&response);
        inbound.extend_from_slice(&server_records(&mut server_upgrade).0);
        let upgraded = block_on(establish(
            Scripted::new(inbound),
            "control.example",
            &identity,
            &server_key,
            &mut Counter(9),
        ))
        .unwrap();
        let mut request = [0; upgrade::MAX_REQUEST];
        let request =
            upgrade::write_upgrade_request("control.example", initiation.as_bytes(), &mut request)
                .unwrap();
        assert_eq!(upgraded.stream.written.as_slice(), request);
        assert_eq!(upgraded.pushback.as_slice(), &settings);
        let mut upgraded = block_on(upgraded.into_http2()).unwrap();

        // The same HTTP/2 preface bytes followed the handshake on both, and
        // both stripped the early payload and kept the server's first frame.
        assert_eq!(
            &upgraded.stream.written[request.len()..],
            split.stream.written.as_slice()
        );
        for http2 in [&split, &upgraded] {
            assert_eq!(
                &http2.plaintext
                    [usize::from(http2.plaintext_start)..usize::from(http2.plaintext_end)],
                &settings
            );
        }

        // Identical record keys and counters: both seal the same bytes, and
        // the server opens them.
        assert_eq!(split.session.counters(), upgraded.session.counters());
        let mut sealed_split = [0; 64];
        let mut sealed_upgraded = [0; 64];
        let len = split.session.seal(b"ping", &mut sealed_split).unwrap();
        assert_eq!(
            upgraded
                .session
                .seal(b"ping", &mut sealed_upgraded)
                .unwrap(),
            len
        );
        assert_eq!(sealed_split[..len], sealed_upgraded[..len]);
        let mut opened = [0; noise::MAX_PLAINTEXT];
        let mut preface = &split.stream.written[..];
        while !preface.is_empty() {
            let len = Session::ciphertext_len(preface).unwrap();
            server_split
                .open(
                    &preface[noise::HEADER_LEN..noise::HEADER_LEN + len],
                    &mut opened,
                )
                .unwrap();
            preface = &preface[noise::HEADER_LEN + len..];
        }
        let opened_len = server_split
            .open(&sealed_split[noise::HEADER_LEN..len], &mut opened)
            .unwrap();
        assert_eq!(&opened[..opened_len], b"ping");
    }

    #[test]
    fn a_websocket_handshake_does_not_wait_for_the_early_payload() {
        // Over WebSocket the server sends only its Noise response, then waits
        // for the client's HTTP/2 preface before anything else.
        let identity = Identity::generate(&mut Counter(1));
        let server_private = StaticSecret::from([0x42; 32]);
        let server_key =
            MachinePublic::from_bytes(x25519_dalek::PublicKey::from(&server_private).to_bytes());
        let (initiation, pending) = initiate(&identity, &server_key, &mut Counter(9)).unwrap();
        let (response, _) = crate::noise::ik::respond_for_test(
            &server_private,
            &StaticSecret::from([0x43; 32]),
            initiation.as_bytes(),
        );
        let stream = block_on(complete(
            Scripted::new(std::vec::Vec::from(response)),
            pending,
        ));
        assert!(stream.is_ok());
    }

    #[test]
    fn a_refused_upgrade_is_reported_before_any_noise_bytes_are_read() {
        let identity = Identity::generate(&mut Counter(1));
        let server_key = MachinePublic::from_bytes([0x42; 32]);
        let inbound = std::vec::Vec::from(&b"HTTP/1.1 403 Forbidden\r\n\r\n"[..]);
        assert!(matches!(
            block_on(establish(
                Scripted::new(inbound),
                "control.example",
                &identity,
                &server_key,
                &mut Counter(9),
            )),
            Err(Error::UnexpectedStatus(403))
        ));
    }

    #[test]
    fn host_storage_is_owned_and_firmware_storage_is_borrowed() {
        struct HostStream;
        type Owned = Http2<HostStream>;
        type Borrowed = Http2<HostStream, &'static mut [u8; H2_BUFFER], &'static mut Connection>;

        assert!(core::mem::size_of::<Owned>() >= H2_BUFFER);
        assert!(core::mem::size_of::<Borrowed>() < core::mem::size_of::<Owned>());
        assert_eq!(H2_BUFFER, 2_048);
    }

    #[test]
    fn map_in_place_stops_at_each_completed_document_and_leaves_the_next_unread() {
        let responses: std::vec::Vec<_> = (1..=4)
            .map(|port| lease_peer_response(7, &std::format!("198.51.100.7:{}", 4000 + port)))
            .collect();
        let bodies: std::vec::Vec<_> = responses.iter().map(|r| lease_map_bytes(r)).collect();
        let splits: std::vec::Vec<_> = responses
            .iter()
            .map(|r| 4 + r.find("\"KeepAlive\"").unwrap())
            .collect();
        let mut first = bodies[0].clone();
        first.extend_from_slice(&bodies[1][..splits[1]]);
        let mut second = bodies[1][splits[1]..].to_vec();
        second.extend_from_slice(&bodies[2][..splits[2]]);
        let mut third = bodies[2][splits[2]..].to_vec();
        third.extend_from_slice(&bodies[3][..splits[3]]);
        let mut fourth = bodies[3][splits[3]..].to_vec();
        fourth.extend_from_slice(&[0; 4]);
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&first),
            lease_data(&second),
            lease_data(&third),
            lease_data(&fourth),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let mut active = lease_netmap::<2>();
        let mut lease = lease_netmap::<2>();
        for completed in 1..=4 {
            assert_eq!(
                block_on(map.next_in_place(&mut http, &mut lease)).unwrap(),
                MapEvent::Applied
            );
            assert!(
                !map.has_partial_response(),
                "the next document must stay unread until the lease is handed off"
            );
            // The firmware node adopts the lease, then reseeds what it returns.
            core::mem::swap(&mut active, &mut lease);
            lease.copy_from(&active);
            assert_eq!(active.responses, completed);
            assert_eq!(
                active.peers.get(7).unwrap().endpoints.as_slice(),
                &[std::format!("198.51.100.7:{}", 4000 + completed)
                    .parse()
                    .unwrap()],
                "Applied exposed peer tokens from the following partial document"
            );
        }
        assert_eq!(
            block_on(map.next_in_place(&mut http, &mut lease)).unwrap(),
            MapEvent::KeepAlive
        );
    }

    #[test]
    fn map_in_place_drained_partial_document_defers_the_coalesced_applied() {
        let response_a = lease_peer_response(7, "198.51.100.7:4007");
        let response_b = lease_peer_response(7, "203.0.113.7:5007");
        let split = response_b.find("\"KeepAlive\"").unwrap();
        let body_b = lease_map_bytes(&response_b);
        let mut combined = lease_map_bytes(&response_a);
        combined.extend_from_slice(&body_b[..4 + split]);
        let mut http = lease_http(&[
            lease_headers(1, false),
            lease_data(&combined),
            lease_headers(3, true),
            lease_data(&body_b[4 + split..]),
            lease_data(&[0; 4]),
        ]);
        let mut map = lease_stream::<2>(&mut http);
        let mut lease = lease_netmap::<2>();
        block_on(send_endpoint_update_in_place(
            &mut http,
            &mut map,
            &mut lease,
            "control.example",
            b"{}",
        ))
        .unwrap();
        assert!(map.has_pending_update());
        assert!(
            map.has_partial_response(),
            "the caller's reseed guard must see that B already lives in the lease"
        );
        assert_eq!(
            block_on(map.next_in_place(&mut http, &mut lease)).unwrap(),
            MapEvent::Applied
        );
        assert!(!map.has_partial_response());
        assert_eq!(lease.responses, 2, "A and B coalesce into one snapshot");
        assert_eq!(
            lease.peers.get(7).unwrap().endpoints.as_slice(),
            &["203.0.113.7:5007".parse().unwrap()]
        );
        assert_eq!(
            block_on(map.next_in_place(&mut http, &mut lease)).unwrap(),
            MapEvent::KeepAlive,
            "the coalesced Applied must not be replayed"
        );
    }
}