// Copyright (c) 2019-2024 Alexander Medvednikov. All rights reserved.
// Use of this source code is governed by an MIT license
// that can be found in the LICENSE file.
module http

import sync
import time

// This file implements a multiplexed HTTP/2 client connection: one connection
// carries many concurrent request streams. A background reader thread (the
// only reader of the transport, and the sole owner of the HPACK decoder and
// read buffer) demuxes incoming frames to per-stream state; request threads
// wait on their stream's condition variable. It complements the synchronous
// single-stream H2Conn in h2_conn.v, which remains for the one-shot paths.
//
// Locking protocol — a thread holds at most one of {smu, fmu, wmu} at a time,
// with two permitted nestings: wmu → smu and smu → fmu (so wmu → smu → fmu).
// No lock is ever taken in the reverse direction, which keeps the order
// acyclic. A stream's own mu is never held while taking a connection lock.
// The smu → fmu nesting is what serializes stream registration (which assigns
// the stream's initial send window) against WINDOW_UPDATE credit and SETTINGS
// initial-window deltas, so early credit can never be lost.
//   wmu      transport writes, the HPACK encoder, stream-id allocation +
//            registration + HEADERS send (one atomic critical section), and
//            the lazy connection preface.
//   fmu/fcv  flow control: the connection send window, every stream's send
//            window, and the peer's initial-window/max-frame mirrors. fcv is
//            broadcast whenever a window can grow and when the connection
//            dies, so blocked senders always wake.
//   smu      connection state: the stream map, refcount, active stream count,
//            goaway/closed/shutting_down, and the peer max-streams mirror.

// h2_max_recv_header_block caps the total size of a received HEADERS(+
// CONTINUATION) block. Without it a peer can stream unbounded CONTINUATION
// frames and exhaust memory (the CONTINUATION-flood DoS, CVE-2024-27316 class).
const h2_max_recv_header_block = 1024 * 1024

// h2_max_pending_preface_acks caps how many control-frame ACKs (SETTINGS + PING)
// may be deferred while waiting to send the lazy client preface. A well-behaved
// server sends only its preface SETTINGS (and perhaps a PING) before ours, so a
// small bound is ample; without it a peer that holds the connection pre-handshake
// could flood PING/SETTINGS and grow the deferred-ACK buffers without limit.
const h2_max_pending_preface_acks = 64

// h2_err_retryable_code tags stream errors where the request provably never
// reached server processing (GOAWAY-unprocessed, connection closed before the
// request was sent, admission refused), so it is safe to retry on a fresh
// connection even for non-idempotent methods.
pub const h2_err_retryable_code = -20012

// h2_retryable_error builds an error carrying h2_err_retryable_code.
fn h2_retryable_error(reason string) IError {
    return error_with_code('h2: ${reason}', h2_err_retryable_code)
}

// all_digits reports whether s is non-empty and every byte is an ASCII digit.
// Header values must be validated with this before lenient string.int()/.u64()
// parsing, which otherwise accept malformed input like '200 OK' or '12junk'.
fn all_digits(s string) bool {
    if s.len == 0 {
        return false
    }
    for ch in s {
        if ch < `0` || ch > `9` {
            return false
        }
    }
    return true
}

// h2_conn_specific_headers are connection-specific header fields that MUST NOT
// appear in any HTTP/2 message (RFC 9113 §8.2.2). A received response or trailer
// carrying one is malformed. (TE is the request-only exception and is handled on
// the send side, so it is not listed here.)
const h2_conn_specific_headers = ['connection', 'keep-alive', 'proxy-connection', 'transfer-encoding',
    'upgrade']

// h2_response_field_error returns a non-empty reason when a regular (non-pseudo)
// received header field is malformed per RFC 9113 §8.2, or '' when it is valid.
// Field names must be non-empty and lowercase (§8.2.1), and connection-specific
// fields are forbidden (§8.2.2). A malformed field makes the whole message
// malformed (§8.1.1); the mux path resets the stream and the sync path fails the
// request rather than delivering it. Pseudo-header validity is checked at the
// call site (the set of valid pseudo-headers differs between headers/trailers).
fn h2_response_field_error(name string) string {
    if name.len == 0 {
        return 'empty header field name'
    }
    for ch in name {
        if ch >= `A` && ch <= `Z` {
            return 'uppercase header field name "${name}"'
        }
    }
    if name in h2_conn_specific_headers {
        return 'connection-specific header field "${name}"'
    }
    return ''
}

// h2_header_list_size returns the RFC 9113 §6.5.2 size of a header list: the sum
// over all fields of (name length + value length + 32 octets of per-field
// overhead). Used to honor the peer's advisory SETTINGS_MAX_HEADER_LIST_SIZE.
fn h2_header_list_size(fields []H2HeaderField) u64 {
    mut total := u64(0)
    for f in fields {
        total += u64(f.name.len) + u64(f.value.len) + 32
    }
    return total
}

// H2MuxStream is the client-side state of one in-flight request stream.
@[heap]
struct H2MuxStream {
mut:
    id u32
    // --- response state, guarded by mu, signaled via cv ---
    mu            &sync.Mutex = unsafe { nil }
    cv            &sync.Cond  = unsafe { nil }
    status        int
    resp_headers  []H2HeaderField
    resp_trailers []H2HeaderField // non-pseudo fields from a trailing HEADERS block
    headers_done  bool
    chunks        [][]u8 // DATA payloads appended by the reader, drained by the requester
    body_rcvd     u64    // cumulative DATA bytes received
    ended         bool   // END_STREAM, RST, or connection death
    err           string // non-empty: the stream failed
    retryable     bool   // the failure is safe to retry on a fresh connection
    sent_headers  bool   // the request HEADERS reached the transport
    cancelled     bool   // requester abandoned the stream; drop+credit late DATA
    send_closed   bool   // we closed our send side (sent END_STREAM or RST_STREAM)
    // --- send state, guarded by the connection's fmu ---
    send_window i64
    send_dead   bool // RST/GOAWAY: wake a body sender blocked on flow-control credit
    // --- receive state, guarded by mu ---
    recv_window i64 = i64(h2_default_initial_window) // our receive window (what we advertised)
}

fn new_h2_mux_stream() &H2MuxStream {
    mu := sync.new_mutex()
    return &H2MuxStream{
        mu: mu
        cv: sync.new_cond(mu)
    }
}

// fail marks the stream failed and wakes its requester.
fn (mut s H2MuxStream) fail(msg string, retryable bool) {
    s.mu.lock()
    if !s.ended {
        s.err = msg
        s.retryable = retryable
        s.ended = true
        s.cv.signal()
    }
    s.mu.unlock()
}

// H2MuxConn is a multiplexed client-side HTTP/2 connection, safe for
// concurrent requests from multiple threads.
@[heap]
pub struct H2MuxConn {
mut:
    transport H2Transport
    // close_transport is called exactly once during teardown and MUST close or
    // interrupt the transport so the reader thread (blocked in transport.read())
    // can exit. It is required (non-nil): the H2Transport interface has no close()
    // of its own, so this is the only way to unblock the reader. new_h2_mux_conn
    // panics on a nil closer.
    close_transport fn () = unsafe { nil }
    // --- guarded by wmu ---
    wmu                       &sync.Mutex = sync.new_mutex()
    encoder                   H2HpackEncoder
    next_stream_id            u32 = 1
    handshaked                bool
    pending_settings_acks     int    // count of SETTINGS frames received before our preface; each needs one ACK
    pending_ping_acks         [][]u8 // PING frames received before our preface; each needs an ACK with the same data
    peer_max_header_list_size u32 = max_u32 // peer SETTINGS_MAX_HEADER_LIST_SIZE (advisory, §6.5.2)
    // --- guarded by fmu, fcv signals growth/death ---
    fmu                 &sync.Mutex = unsafe { nil }
    fcv                 &sync.Cond  = unsafe { nil }
    send_window         i64         = i64(h2_default_initial_window)
    peer_initial_window i64         = i64(h2_default_initial_window)
    peer_max_frame      u32         = h2_default_max_frame_size
    fmu_dead            bool // mirror of `closed` for blocked senders
    // --- guarded by smu ---
    smu                 &sync.Mutex = sync.new_mutex()
    streams             map[u32]&H2MuxStream
    refs                int = 1 // the owner's (pool's) reference; +1 per in-flight request
    active_streams      int
    max_streams         u32 = 100 // our own cap on concurrent streams
    peer_max_streams    u32 = max_u32
    goaway              bool
    goaway_last         u32
    closed              bool // the reader has exited; only the reader sets this
    shutting_down       bool
    transport_torn_down bool // close_transport has been (or is being) called
    conn_err            string
    idle_since          time.Time
    // --- guarded by recv_wmu ---
    // Inbound flow-control: tracks how much more data the peer is allowed to
    // send us (our advertised receive windows). Debited on DATA arrival,
    // replenished when we send WINDOW_UPDATE. recv_wmu is always acquired
    // solo — never while holding any other connection lock.
    recv_wmu         &sync.Mutex = sync.new_mutex()
    conn_recv_window i64         = i64(h2_default_initial_window)
    // --- reader-thread private (no locks) ---
    decoder H2HpackDecoder
    rbuf    []u8
}

// new_h2_mux_conn creates a multiplexed connection over `transport` and starts
// its background reader. The connection preface is sent lazily with the first
// request. `close_transport` is REQUIRED (non-nil): it is called once at final
// teardown and MUST close/interrupt the transport so the blocked reader exits.
// A nil closer is a programming error and panics here, because the reader can
// only be unblocked by closing the transport (see the field comment).
//
// CONCURRENCY REQUIREMENT: the background reader calls `transport.read()` while
// request threads call `transport.write()`, so `transport` MUST be safe for a
// read and a write to run simultaneously on separate threads. A raw TCP socket
// is (it is full-duplex); a TLS connection is NOT — one OpenSSL/mbedTLS `SSL`
// object must not be read and written at the same time. Wiring this to a TLS
// backend therefore requires a concurrent-safe transport adapter (a serialized
// or non-blocking I/O loop); a single blocking I/O mutex is insufficient because
// it would deadlock the blocked reader. That adapter is provided when the pooled
// transport is wired up (see the HTTP connection-pooling Phase 3 work); until
// then this type is exercised only over a full-duplex in-memory pipe in tests.
pub fn new_h2_mux_conn(transport H2Transport, close_transport fn ()) &H2MuxConn {
    if close_transport == unsafe { nil } {
        // The reader thread spawned below blocks in transport.read(); the
        // H2Transport interface has no close(), so this callback is the only way
        // to interrupt that read at teardown. Without it an idle retirement or
        // last-reference release would leak the reader thread and the connection.
        panic('new_h2_mux_conn: close_transport must be non-nil so teardown can wake the reader')
    }
    fmu := sync.new_mutex()
    mut c := &H2MuxConn{
        transport:       transport
        close_transport: close_transport
        fmu:             fmu
        fcv:             sync.new_cond(fmu)
        idle_since:      time.now()
    }
    spawn c.read_loop()
    return c
}

// can_take_new_request reports whether a new request may be admitted on this
// connection right now (it may still be refused later under smu).
pub fn (mut c H2MuxConn) can_take_new_request() bool {
    c.smu.lock()
    defer {
        c.smu.unlock()
    }
    limit := if c.peer_max_streams < c.max_streams { c.peer_max_streams } else { c.max_streams }
    return !c.closed && !c.goaway && !c.shutting_down && u32(c.active_streams) < limit
}

// shutdown_when_idle asks the connection to retire: no new requests are
// admitted, and once no requests are in flight and the owner's reference is
// released, the transport is closed.
pub fn (mut c H2MuxConn) shutdown_when_idle() {
    c.smu.lock()
    c.shutting_down = true
    c.smu.unlock()
}

// release drops the owner's reference. The connection tears its transport
// down once the reader has exited and all references are gone.
pub fn (mut c H2MuxConn) release() {
    c.drop_ref()
}

fn (mut c H2MuxConn) drop_ref() {
    c.smu.lock()
    c.refs--
    last_ref := c.refs <= 0
    c.smu.unlock()
    if last_ref {
        // Tear the transport down even if the reader has not exited yet: closing
        // it interrupts the reader's blocking read so it can exit. Otherwise an
        // idle connection that is shut down and released leaks the reader thread
        // and the socket, because the reader only sets `closed` after a read
        // error/timeout, which never arrives on a silent transport.
        c.teardown_transport()
    }
}

// teardown_transport runs the close_transport callback exactly once. Calling it
// while the reader is still blocked in transport.read() interrupts that read, so
// the reader observes the closed transport and exits via fail_conn.
fn (mut c H2MuxConn) teardown_transport() {
    c.smu.lock()
    already := c.transport_torn_down
    c.transport_torn_down = true
    c.smu.unlock()
    if !already && c.close_transport != unsafe { nil } {
        c.close_transport()
    }
}

// --- request side -----------------------------------------------------------

// do sends one request over the connection, concurrently with other streams,
// and returns its response. Errors carrying h2_err_retryable_code are safe to
// retry on a fresh connection.
pub fn (mut c H2MuxConn) do(req H2ClientRequest) !H2ClientResponse {
    // Admission.
    c.smu.lock()
    if c.closed {
        reason := if c.conn_err != '' { c.conn_err } else { 'connection is closed' }
        c.smu.unlock()
        return h2_retryable_error(reason)
    }
    if c.goaway || c.shutting_down {
        c.smu.unlock()
        return h2_retryable_error('connection is shutting down')
    }
    limit := if c.peer_max_streams < c.max_streams { c.peer_max_streams } else { c.max_streams }
    if u32(c.active_streams) >= limit {
        c.smu.unlock()
        return h2_retryable_error('connection is at its concurrent stream limit')
    }
    c.refs++
    c.active_streams++
    c.smu.unlock()

    mut s := new_h2_mux_stream()
    resp := c.do_on_stream(mut s, req) or {
        c.finish_stream(mut s)
        return err
    }
    c.finish_stream(mut s)
    return resp
}

// finish_stream removes the stream from the connection and releases the
// request's reference (possibly triggering teardown).
fn (mut c H2MuxConn) finish_stream(mut s H2MuxStream) {
    c.smu.lock()
    c.streams.delete(s.id)
    c.active_streams--
    c.idle_since = time.now()
    c.smu.unlock()
    // Credit any DATA bytes that were queued in s.chunks but never drained.
    // On the success path wait_response empties chunks before returning, so
    // queued is 0. On the error path (do_on_stream returned early, e.g. after
    // a RST_STREAM arrived mid-upload), chunks may hold bytes whose flow-size
    // was already debited from conn_recv_window; without this credit those bytes
    // are silently dropped, permanently shrinking the connection receive window.
    // Setting cancelled prevents a DATA frame in flight from re-queuing onto
    // the now-deregistered stream and leaking more bytes.
    s.mu.lock()
    s.cancelled = true
    mut queued := u64(0)
    for ch in s.chunks {
        queued += u64(ch.len)
    }
    s.chunks.clear()
    s.mu.unlock()
    if queued > 0 {
        c.send_conn_window_update(u32(queued)) or { c.note_write_failure() }
    }
    c.drop_ref()
}

fn (mut c H2MuxConn) do_on_stream(mut s H2MuxStream, req H2ClientRequest) !H2ClientResponse {
    mut fields := [
        H2HeaderField{':method', req.method},
        H2HeaderField{':scheme', req.scheme},
        H2HeaderField{':authority', req.authority},
        H2HeaderField{':path', req.path},
    ]
    for h in req.headers {
        fields << h
    }
    has_body := req.body.len > 0

    // Stream-id allocation, registration, HPACK encoding and the HEADERS(+
    // CONTINUATION) send form one wmu critical section: ids must hit the wire
    // in increasing order, header blocks must be contiguous, and the stream
    // must be registered before its first byte is sent so the reader can
    // always deliver the response.
    c.wmu.lock()
    c.handshake_locked() or {
        // Preface write failed: tear down the transport so the reader exits
        // and the pool stops admitting new work to this dead connection.
        // Without this, closed/shutting_down stay false and the pool can keep
        // dispatching to a connection whose write side is already broken.
        c.wmu.unlock()
        c.note_write_failure()
        return h2_retryable_error('connection handshake failed: ${err.msg()}')
    }
    // RFC 9113 §6.5.2: honor the peer's advisory SETTINGS_MAX_HEADER_LIST_SIZE.
    // Refuse an over-limit request here rather than emit it and have the server
    // reject it (e.g. 431). Not retryable: a fresh connection to the same peer
    // carries the same limit. Read under wmu (apply_peer_settings sets it there).
    peer_max_list := c.peer_max_header_list_size
    if peer_max_list != max_u32 && h2_header_list_size(fields) > u64(peer_max_list) {
        c.wmu.unlock()
        return error('h2: request header list (${h2_header_list_size(fields)} bytes) exceeds peer SETTINGS_MAX_HEADER_LIST_SIZE (${peer_max_list})')
    }
    if c.next_stream_id > u32(0x7fff_ffff) {
        // RFC 7540 §5.1.1: client stream IDs are odd and must not exceed 2^31-1.
        // Retire this connection and let the caller open a fresh one.
        c.smu.lock()
        c.shutting_down = true
        c.smu.unlock()
        c.wmu.unlock()
        return h2_retryable_error('stream ID space exhausted')
    }
    s.id = c.next_stream_id
    c.next_stream_id += 2
    // Mark the HEADERS as sent *before* the stream becomes visible to the
    // reader: pessimistic, so a connection death racing this section can never
    // classify a request whose HEADERS may have reached the wire as safe to
    // replay. (The stream is still private here, so no lock is needed.)
    s.sent_headers = true
    c.smu.lock() // wmu -> smu -> fmu is the permitted lock nesting
    // A terminal event (GOAWAY, or fail_conn after the reader saw the transport
    // die) can land while we are blocked on wmu, after do()'s admission check
    // already passed. Such an event fails every stream in the map, but this one
    // is not registered yet, so it would slip through and wait_response could
    // block forever. Recheck under smu — the same lock those events use to set
    // these flags — and abort before registering or sending. The HEADERS never
    // hit the wire here, so the request is safe to retry on a fresh connection.
    // Also recheck the peer's stream limit: SETTINGS_MAX_CONCURRENT_STREAMS can
    // arrive while we are blocked on wmu, lowering the limit below active_streams.
    if c.closed || c.goaway || c.shutting_down {
        reason := if c.conn_err != '' { c.conn_err } else { 'connection is shutting down' }
        c.smu.unlock()
        c.wmu.unlock()
        return h2_retryable_error(reason)
    }
    recheck_limit := if c.peer_max_streams < c.max_streams {
        c.peer_max_streams
    } else {
        c.max_streams
    }
    if u32(c.active_streams) > recheck_limit {
        c.smu.unlock()
        c.wmu.unlock()
        return h2_retryable_error('peer lowered concurrent stream limit; retrying on a fresh connection')
    }
    // If this was the last valid client stream ID (RFC 7540 §5.1.1: max 2^31-1),
    // retire the connection so can_take_new_request() returns false immediately.
    // Without this, the pool dispatches one extra request that hits the admission
    // check at the top of do_on_stream and fails with a retryable error.
    if c.next_stream_id > u32(0x7fff_ffff) {
        c.shutting_down = true
    }
    c.fmu.lock()
    // The initial send window must be assigned atomically with registration:
    // the WINDOW_UPDATE and SETTINGS handlers also nest smu -> fmu, so credit
    // or a delta arriving right after our HEADERS serializes after this
    // assignment instead of being overwritten by it.
    s.send_window = c.peer_initial_window
    c.streams[s.id] = s
    c.fmu.unlock()
    c.smu.unlock()
    block := c.encoder.encode(fields)
    c.send_header_block_locked(s.id, block, !has_body) or {
        c.wmu.unlock()
        c.note_write_failure()
        // The HEADERS may have partially hit the wire; not safe to blind-retry
        // unless the transport wrote nothing, which we cannot distinguish here.
        return error('h2: failed to send request headers: ${err.msg()}')
    }
    c.wmu.unlock()

    if has_body {
        c.send_body_on_stream(mut s, req.body)!
    } else {
        // The HEADERS carried END_STREAM, so our send side is now closed.
        s.mu.lock()
        s.send_closed = true
        s.mu.unlock()
    }
    return c.wait_response(mut s, req)
}

// handshake_locked sends the connection preface and our SETTINGS once.
// Callers must hold wmu.
fn (mut c H2MuxConn) handshake_locked() ! {
    if c.handshaked {
        return
    }
    mut buf := h2_client_preface.bytes()
    buf << H2Frame(H2SettingsFrame{
        settings: [
            H2Setting{h2_settings_enable_push, 0},
            H2Setting{h2_settings_initial_window_size, h2_default_initial_window},
            H2Setting{h2_settings_max_frame_size, h2_default_max_frame_size},
        ]
    }).encode()
    c.write_all_locked(buf)!
    c.handshaked = true
    // If the reader processed server frames before we sent our preface, ACKs were
    // deferred to preserve client-preface-first ordering (RFC 7540 §3.5). Flush
    // them now: one SETTINGS ACK per received non-ACK SETTINGS, then any PING ACKs.
    for _ in 0 .. c.pending_settings_acks {
        c.write_all_locked(H2Frame(H2SettingsFrame{ ack: true }).encode())!
    }
    c.pending_settings_acks = 0
    for data in c.pending_ping_acks {
        c.write_all_locked(H2Frame(H2PingFrame{ ack: true, data: data }).encode())!
    }
    c.pending_ping_acks.clear()
}

// send_header_block_locked writes a header block as HEADERS(+CONTINUATION)
// frames. Callers must hold wmu.
fn (mut c H2MuxConn) send_header_block_locked(stream_id u32, block []u8, end_stream bool) ! {
    c.fmu.lock()
    mut max := int(c.peer_max_frame)
    c.fmu.unlock()
    if block.len <= max {
        // Re-read peer_max_frame right before writing to close the TOCTOU window:
        // the peer may have lowered SETTINGS_MAX_FRAME_SIZE since the check above.
        c.fmu.lock()
        max = int(c.peer_max_frame)
        c.fmu.unlock()
        if block.len <= max {
            c.write_all_locked(H2Frame(H2HeadersFrame{
                stream_id:   stream_id
                fragment:    block
                end_headers: true
                end_stream:  end_stream
            }).encode())!
            return
        }
        // block no longer fits in one frame under the refreshed limit;
        // fall through to the multi-frame path.
    }
    // Re-read peer_max_frame under fmu immediately before the first HEADERS write
    // to minimise the TOCTOU window between the size-check above and this write.
    c.fmu.lock()
    max = int(c.peer_max_frame)
    c.fmu.unlock()
    // Clamp to block.len in case max grew large enough to fit the whole block.
    first := if max < block.len { max } else { block.len }
    c.write_all_locked(H2Frame(H2HeadersFrame{
        stream_id:   stream_id
        fragment:    block[..first]
        end_headers: first == block.len
        end_stream:  end_stream
    }).encode())!
    if first == block.len {
        return
    }
    mut off := first
    for off < block.len {
        // Re-read peer_max_frame under fmu before each CONTINUATION to minimise
        // the TOCTOU window: the peer could send a smaller SETTINGS_MAX_FRAME_SIZE
        // between iterations and enforce it on the next frame we send.
        c.fmu.lock()
        cur_max := int(c.peer_max_frame)
        c.fmu.unlock()
        mut next := off + cur_max
        if next > block.len {
            next = block.len
        }
        c.write_all_locked(H2Frame(H2ContinuationFrame{
            stream_id:   stream_id
            fragment:    block[off..next]
            end_headers: next == block.len
        }).encode())!
        off = next
    }
}

// send_body_on_stream writes the request body as DATA frames, bounded by both
// the connection and stream send windows.
fn (mut c H2MuxConn) send_body_on_stream(mut s H2MuxStream, body []u8) ! {
    mut off := 0
    for off < body.len {
        // Reserve a window-bounded chunk under fmu (waiting for WINDOW_UPDATE
        // room when both windows are exhausted), then write it under wmu.
        c.fmu.lock()
        for !c.fmu_dead && !s.send_dead && (c.send_window <= 0 || s.send_window <= 0) {
            c.fcv.wait()
        }
        if c.fmu_dead || s.send_dead {
            stream_dead := s.send_dead
            c.fmu.unlock()
            if stream_dead {
                // send_dead is always set together with a terminal stream state,
                // so distinguish the two by whether fail() recorded an error:
                //  - no error  → the server sent END_STREAM (early final response)
                //    while we still owed body; abandon the upload and close our
                //    half (below) so wait_response can still deliver the response.
                //  - error set → RST_STREAM, GOAWAY, or connection death; surface
                //    it here, preserving retryability so a request the server never
                //    processed can be replayed on a fresh connection.
                s.mu.lock()
                failure := s.err
                failure_retryable := s.retryable
                s.mu.unlock()
                if failure != '' {
                    if failure_retryable {
                        return h2_retryable_error(failure)
                    }
                    return error('h2: ${failure}')
                }
                // Early final response: the server completed and closed its half
                // while we still owed body. RST_STREAM(CANCEL) closes our half so
                // the server releases the stream instead of holding it half-open
                // and counting it against its concurrency limit (RFC 9113 §8.1).
                // The stream stays registered in c.streams, so wait_response still
                // returns the already-received response from the s reference.
                c.wmu.lock()
                c.write_all_locked(H2Frame(H2RstStreamFrame{
                    stream_id:  s.id
                    error_code: u32(H2ErrorCode.cancel)
                }).encode()) or {
                    c.wmu.unlock()
                    c.note_write_failure()
                    return
                }
                c.wmu.unlock()
                s.mu.lock()
                s.send_closed = true
                s.mu.unlock()
                return
            }
            return error('h2: connection closed while sending the request body')
        }
        mut chunk := body.len - off
        if chunk > int(c.peer_max_frame) {
            chunk = int(c.peer_max_frame)
        }
        if i64(chunk) > c.send_window {
            chunk = int(c.send_window)
        }
        if i64(chunk) > s.send_window {
            chunk = int(s.send_window)
        }
        c.send_window -= i64(chunk)
        s.send_window -= i64(chunk)
        c.fmu.unlock()

        mut next := off + chunk
        c.wmu.lock()
        // Re-cap chunk under wmu→fmu (permitted order): the reader may have
        // processed SETTINGS_MAX_FRAME_SIZE and sent the ACK between our
        // fmu.unlock() above and wmu.lock() here; re-read and return any excess.
        c.fmu.lock()
        if chunk > int(c.peer_max_frame) {
            excess := i64(chunk) - i64(c.peer_max_frame)
            c.send_window += excess
            s.send_window += excess
            chunk = int(c.peer_max_frame)
            next = off + chunk
            // Wake any goroutines sleeping in fcv.wait() because c.send_window
            // was drained; no WINDOW_UPDATE is coming for these bytes since they
            // were never sent.
            c.fcv.broadcast()
        }
        // Also revalidate stream send window: an INITIAL_WINDOW_SIZE reduction
        // applies a negative delta to s.send_window under fmu, so a negative
        // value here means we over-claimed against the peer's new window.  Return
        // the excess bytes to both windows so the loop can re-wait for capacity.
        if s.send_window < 0 {
            trim := if i64(chunk) < -s.send_window { i64(chunk) } else { -s.send_window }
            chunk -= int(trim)
            s.send_window += trim
            c.send_window += trim
            next = off + chunk
            c.fcv.broadcast()
        }
        c.fmu.unlock()
        if chunk == 0 {
            c.wmu.unlock()
            continue
        }
        c.write_all_locked(H2Frame(H2DataFrame{
            stream_id:  s.id
            data:       body[off..next]
            end_stream: next == body.len
        }).encode()) or {
            c.wmu.unlock()
            c.note_write_failure()
            return error('h2: failed to send request body: ${err.msg()}')
        }
        c.wmu.unlock()
        off = next
    }
    // The final DATA frame carried END_STREAM, so our send side is now closed.
    s.mu.lock()
    s.send_closed = true
    s.mu.unlock()
}

// wait_response drains the stream until it ends, honoring the request's
// streaming callback and stop limits. Callbacks run on the requester's thread.
fn (mut c H2MuxConn) wait_response(mut s H2MuxStream, req H2ClientRequest) !H2ClientResponse {
    mut resp := H2ClientResponse{}
    mut body_expected := u64(0)
    mut has_content_length := false
    mut got_headers := false
    mut body_so_far := u64(0)
    mut cancelled := false
    s.mu.lock()
    for {
        // Drain everything currently buffered.
        if !got_headers && s.headers_done {
            resp.status = s.status
            for f in s.resp_headers {
                resp.headers << f
                // on_response_headers already rejects a non-numeric Content-Length,
                // but guard the lenient u64() at the parse site too so this stays
                // correct even if that upstream check is ever changed.
                if f.name == 'content-length' && all_digits(f.value) {
                    body_expected = f.value.u64()
                    has_content_length = true
                }
            }
            got_headers = true
        }
        mut drained := u64(0)
        for s.chunks.len > 0 {
            chunk := s.chunks[0]
            s.chunks.delete(0)
            body_so_far += u64(chunk.len)
            drained += u64(chunk.len)
            if req.stop_copying_limit < 0
                || i64(body_so_far) - i64(chunk.len) < req.stop_copying_limit {
                if req.stop_copying_limit >= 0 && i64(body_so_far) > req.stop_copying_limit {
                    remaining := req.stop_copying_limit - (i64(body_so_far) - i64(chunk.len))
                    if remaining > 0 {
                        resp.body << chunk[..int(remaining)]
                    }
                } else {
                    resp.body << chunk
                }
            }
            if req.on_data != unsafe { nil } {
                // Run the user callback outside the stream lock so it can
                // block without stalling the reader's delivery.
                s.mu.unlock()
                req.on_data(chunk, body_so_far, body_expected, resp.status) or {
                    // The callback aborted the request: credit the connection
                    // window for what we consumed this round and RST the stream,
                    // just like the stop_receiving_limit path, so the connection
                    // window does not leak and the peer stops sending.
                    if drained > 0 {
                        c.send_conn_window_update(u32(drained)) or {}
                    }
                    c.cancel_stream(mut s)
                    return err
                }
                s.mu.lock()
            }
            if req.stop_receiving_limit >= 0 && i64(body_so_far) >= req.stop_receiving_limit {
                cancelled = true
                break
            }
        }
        ended := s.ended
        serr := s.err
        retryable := s.retryable
        if cancelled {
            s.mu.unlock()
            // The connection-level window must still be credited for the bytes
            // this round consumed, or the connection's receive window shrinks
            // permanently for every other stream.
            if drained > 0 {
                c.send_conn_window_update(u32(drained)) or {}
            }
            c.cancel_stream(mut s)
            if !got_headers {
                return error('h2: stream cancelled before a response arrived')
            }
            return resp
        }
        if drained > 0 {
            // Replenish flow control for what was just consumed, outside s.mu.
            s.mu.unlock()
            c.send_window_updates(s.id, u32(drained)) or {}
            s.mu.lock()
            // New chunks may have arrived while unlocked; loop and re-drain.
            if s.chunks.len > 0 || (s.ended && !ended) {
                continue
            }
        }
        if ended {
            // Surface any trailers (a second HEADERS block) into the response
            // headers, mirroring the synchronous H2Conn.read_response path. The
            // stream has ended, so resp_trailers is complete; copy under the lock.
            for f in s.resp_trailers {
                resp.headers << f
            }
            s.mu.unlock()
            if serr != '' {
                if retryable {
                    return h2_retryable_error(serr)
                }
                return error('h2: ${serr}')
            }
            if !got_headers {
                return error('h2: stream closed without a response')
            }
            // RFC 9110 §8.6: a Content-Length must match the bytes received.
            // Skip for responses defined to carry no body — HEAD requests and
            // 204/304 status codes — where Content-Length describes the absent
            // representation rather than transmitted DATA.
            body_allowed := req.method != 'HEAD' && resp.status != 204 && resp.status != 304
            if has_content_length && body_allowed && body_so_far != body_expected {
                return error('h2: response body length ${body_so_far} does not match Content-Length ${body_expected}')
            }
            return resp
        }
        s.cv.wait()
    }
    return resp
}

// cancel_stream aborts a stream early (stop_receiving_limit): the stream is
// deregistered first and then RST_STREAM is sent, so any in-flight late DATA
// for it is handled by the reader's unknown-stream backstop (connection-level
// WINDOW_UPDATE), keeping the connection fully usable for other streams.
fn (mut c H2MuxConn) cancel_stream(mut s H2MuxStream) {
    // Deregister first, then send RST_STREAM: once the stream is gone from
    // c.streams, any DATA the reader already had in flight for it hits the
    // unknown-stream backstop, which credits the connection-level receive
    // window. If we sent RST first, that in-flight DATA could still find the
    // registered stream and be queued without a WINDOW_UPDATE (the cancelled
    // requester has stopped draining), permanently leaking connection window.
    c.smu.lock()
    c.streams.delete(s.id)
    c.smu.unlock()
    // DATA already queued before cancellation consumed the connection receive
    // window but will never be drained; credit it back (its padding was already
    // credited on receipt, so only the chunk data remains). Setting `cancelled`
    // under s.mu also makes a frame still in flight credit-and-drop in
    // on_response_data instead of queuing onto a stream nobody drains. Without
    // this, repeated early cancellations permanently shrink the connection window.
    s.mu.lock()
    s.cancelled = true
    mut queued := u64(0)
    for ch in s.chunks {
        queued += u64(ch.len)
    }
    s.chunks.clear()
    s.mu.unlock()
    if queued > 0 {
        // A write failure here means the transport is already dead; tear it down
        // immediately so the pool stops reusing it. note_write_failure is
        // sufficient — the RST_STREAM is skipped since the peer will not receive
        // it on a dead transport.
        c.send_conn_window_update(u32(queued)) or {
            c.note_write_failure()
            return
        }
    }
    c.wmu.lock()
    c.write_all_locked(H2Frame(H2RstStreamFrame{
        stream_id:  s.id
        error_code: u32(H2ErrorCode.cancel)
    }).encode()) or {
        c.wmu.unlock()
        c.note_write_failure()
        return
    }
    c.wmu.unlock()
}

// send_conn_window_update replenishes only the connection-level receive
// window (used when the stream itself is being cancelled).
fn (mut c H2MuxConn) send_conn_window_update(n u32) ! {
    if n == 0 {
        return
    }
    // Credit our tracked window before sending the frame so the budget is
    // never in deficit for longer than necessary.
    c.recv_wmu.lock()
    c.conn_recv_window += i64(n)
    c.recv_wmu.unlock()
    buf := H2Frame(H2WindowUpdateFrame{
        stream_id:             0
        window_size_increment: n
    }).encode()
    c.wmu.lock()
    c.write_all_locked(buf) or {
        c.wmu.unlock()
        c.note_write_failure()
        return err
    }
    c.wmu.unlock()
}

// send_window_updates replenishes both the connection and stream receive
// windows after the requester consumed `n` body bytes.
fn (mut c H2MuxConn) send_window_updates(stream_id u32, n u32) ! {
    if n == 0 {
        return
    }
    // Credit both tracked receive windows before sending the frames.
    c.recv_wmu.lock()
    c.conn_recv_window += i64(n)
    c.recv_wmu.unlock()
    mut stream_alive := false
    c.smu.lock()
    if mut s := c.streams[stream_id] {
        s.mu.lock()
        s.recv_window += i64(n)
        s.mu.unlock()
        stream_alive = true
    }
    c.smu.unlock()
    mut buf := H2Frame(H2WindowUpdateFrame{
        stream_id:             0
        window_size_increment: n
    }).encode()
    if stream_alive {
        buf << H2Frame(H2WindowUpdateFrame{
            stream_id:             stream_id
            window_size_increment: n
        }).encode()
    }
    c.wmu.lock()
    c.write_all_locked(buf) or {
        c.wmu.unlock()
        c.note_write_failure()
        return err
    }
    c.wmu.unlock()
}

// write_all_locked writes all of `data` to the transport. Callers hold wmu.
fn (mut c H2MuxConn) write_all_locked(data []u8) ! {
    mut sent := 0
    for sent < data.len {
        n := c.transport.write(data[sent..])!
        if n <= 0 {
            return error('transport write returned ${n}')
        }
        sent += n
    }
}

// note_write_failure handles a transport write failure: it stops admitting new
// requests and tears the transport down. Closing it interrupts the reader's
// blocking read so it runs fail_conn and fails every other in-flight stream,
// instead of leaving them hung when the transport's read side does not also
// break (a write-only failure). teardown_transport is once-guarded, so this
// never double-closes against the reader's own teardown.
fn (mut c H2MuxConn) note_write_failure() {
    c.smu.lock()
    c.shutting_down = true
    c.smu.unlock()
    // Interrupt the reader's blocking read if a closer can, then fail the
    // connection directly. fail_conn is idempotent (guarded by c.closed), so a
    // later reader invocation is a safe no-op. Calling it unconditionally — not
    // only when close_transport is nil — wakes every in-flight stream
    // immediately, closing both the window before the reader notices the dead
    // transport and the case of a closer that cannot interrupt a blocking read.
    c.teardown_transport()
    c.fail_conn('transport write failure')
}

// --- reader side -------------------------------------------------------------

// read_loop runs on the connection's background thread: it is the only reader
// of the transport and demuxes every incoming frame.
fn (mut c H2MuxConn) read_loop() {
    for {
        frame := c.mux_read_frame() or {
            if is_transport_timeout_error(err) {
                if c.reader_should_exit() {
                    c.fail_conn('connection retired while idle')
                    return
                }
                continue
            }
            c.fail_conn('connection lost: ${err.msg()}')
            return
        }
        c.dispatch_frame(frame) or {
            c.fail_conn(err.msg())
            return
        }
    }
}

// reader_should_exit lets an idle reader retire the connection on shutdown.
fn (mut c H2MuxConn) reader_should_exit() bool {
    c.smu.lock()
    defer {
        c.smu.unlock()
    }
    return c.shutting_down && c.active_streams == 0
}

// is_transport_timeout_error recognizes read-timeout errors, which wake the
// reader for shutdown checks rather than killing the connection.
fn is_transport_timeout_error(err IError) bool {
    msg := err.msg().to_lower()
    return msg.contains('timed out') || msg.contains('timeout')
}

// mux_read_frame reads and decodes one frame from the transport, enforcing
// the receive limit we advertised to the peer in our own SETTINGS. This is
// h2_default_max_frame_size, which H2MuxConn always sends and never renegotiates.
// (c.peer_max_frame is the peer's receive limit — our outbound cap — and must
// not be used here.)
fn (mut c H2MuxConn) mux_read_frame() !H2Frame {
    c.mux_fill_at_least(h2_frame_header_len)!
    header := h2_parse_frame_header(c.rbuf)!
    if header.length > h2_default_max_frame_size {
        return error('frame larger than SETTINGS_MAX_FRAME_SIZE (${header.length})')
    }
    total := h2_frame_header_len + int(header.length)
    c.mux_fill_at_least(total)!
    frame := h2_parse_frame(header, c.rbuf[h2_frame_header_len..total])!
    c.rbuf = c.rbuf[total..].clone()
    return frame
}

// mux_fill_at_least reads from the transport until rbuf holds n bytes.
fn (mut c H2MuxConn) mux_fill_at_least(n int) ! {
    for c.rbuf.len < n {
        mut tmp := []u8{len: h2_conn_read_chunk}
        got := c.transport.read(mut tmp)!
        if got <= 0 {
            return error('connection closed by peer')
        }
        c.rbuf << tmp[..got]
    }
}

fn (mut c H2MuxConn) dispatch_frame(frame H2Frame) ! {
    match frame {
        H2SettingsFrame {
            if !frame.ack {
                c.apply_peer_settings(frame.settings)!
                c.wmu.lock()
                if !c.handshaked {
                    // RFC 7540 §3.5: no frame may precede the client connection
                    // preface. Defer the ACK; handshake_locked() will flush it
                    // immediately after sending the preface. Use a counter so
                    // multiple SETTINGS frames each get their own ACK.
                    c.pending_settings_acks++
                    over := c.pending_settings_acks + c.pending_ping_acks.len > h2_max_pending_preface_acks
                    c.wmu.unlock()
                    if over {
                        return error('h2: too many control frames before the client preface')
                    }
                } else {
                    c.write_all_locked(H2Frame(H2SettingsFrame{
                        ack: true
                    }).encode()) or {
                        c.wmu.unlock()
                        return error('failed to ack SETTINGS: ${err.msg()}')
                    }
                    c.wmu.unlock()
                }
            }
        }
        H2PingFrame {
            if !frame.ack {
                c.wmu.lock()
                if !c.handshaked {
                    // RFC 7540 §3.5: client preface must be the first bytes sent.
                    // Defer the ACK; handshake_locked() will flush it after the preface.
                    c.pending_ping_acks << frame.data
                    over := c.pending_settings_acks + c.pending_ping_acks.len > h2_max_pending_preface_acks
                    c.wmu.unlock()
                    if over {
                        return error('h2: too many control frames before the client preface')
                    }
                } else {
                    c.write_all_locked(H2Frame(H2PingFrame{
                        ack:  true
                        data: frame.data
                    }).encode()) or {
                        c.wmu.unlock()
                        return error('failed to ack PING: ${err.msg()}')
                    }
                    c.wmu.unlock()
                }
            }
        }
        H2WindowUpdateFrame {
            inc := frame.window_size_increment
            if frame.stream_id == 0 {
                // RFC 7540 6.9: a 0 increment is a connection PROTOCOL_ERROR;
                // a window past 2^31-1 is a connection FLOW_CONTROL_ERROR.
                if inc == 0 {
                    return error('h2: connection WINDOW_UPDATE with a zero increment')
                }
                c.fmu.lock()
                new_window := c.send_window + i64(inc)
                overflow := new_window > i64(0x7fff_ffff)
                if !overflow {
                    c.send_window = new_window
                    c.fcv.broadcast()
                }
                c.fmu.unlock()
                if overflow {
                    return error('h2: connection flow-control window exceeded 2^31-1')
                }
            } else {
                // Stream-level versions of the same rules are stream errors:
                // RST_STREAM the offending stream instead of killing the conn.
                if inc == 0 {
                    c.reset_stream(frame.stream_id, .protocol_error,
                        'WINDOW_UPDATE with a zero increment')
                    return
                }
                // Hold smu across the lookup and the credit (smu -> fmu), so
                // this serializes with stream registration and the credit can
                // never be overwritten by the initial-window assignment.
                c.smu.lock()
                mut sref := c.streams[frame.stream_id] or { &H2MuxStream(unsafe { nil }) }
                mut overflow := false
                if sref != unsafe { nil } {
                    c.fmu.lock()
                    new_window := sref.send_window + i64(inc)
                    overflow = new_window > i64(0x7fff_ffff)
                    if !overflow {
                        sref.send_window = new_window
                        c.fcv.broadcast()
                    }
                    c.fmu.unlock()
                }
                c.smu.unlock()
                if overflow {
                    c.reset_stream(frame.stream_id, .flow_control_error,
                        'stream flow-control window exceeded 2^31-1')
                }
            }
        }
        H2GoawayFrame {
            // Take wmu before smu (the permitted wmu -> smu nesting) so setting
            // c.goaway serializes with do_on_stream's terminal-flag recheck, which
            // runs under smu while holding wmu. Without this, a GOAWAY landing
            // between that recheck (smu released at the end of the registration
            // section) and the HEADERS write (still under wmu) lets the client
            // open one more stream after observing GOAWAY (RFC 7540 6.8). Holding
            // wmu here means c.goaway cannot be set while any writer is mid-section,
            // so the recheck is authoritative.
            c.wmu.lock()
            c.smu.lock()
            c.goaway = true
            c.goaway_last = frame.last_stream_id
            mut above := []&H2MuxStream{}
            for id, st in c.streams {
                if id > frame.last_stream_id {
                    above << st
                }
            }
            c.smu.unlock()
            c.wmu.unlock()
            for mut st in above {
                // Streams above last_stream_id were not processed by the
                // server, so they are safe to retry elsewhere (RFC 7540 6.8).
                st.fail('request not processed (GOAWAY)', true)
                c.wake_send(mut st)
            }
            if frame.error_code != u32(H2ErrorCode.no_error) {
                return error('connection error (GOAWAY ${h2_error_code_name(frame.error_code)})')
            }
        }
        H2HeadersFrame {
            c.on_response_headers(frame)!
        }
        H2DataFrame {
            c.on_response_data(frame)!
        }
        H2RstStreamFrame {
            mut s := c.lookup_stream(frame.stream_id)
            if s != unsafe { nil } {
                // REFUSED_STREAM means the server did not process the request
                // (RFC 7540 8.1.4), so it is safe to replay on a fresh connection
                // even for a non-idempotent method; any other reset code is not.
                retryable := frame.error_code == u32(H2ErrorCode.refused_stream)
                s.fail('stream reset by peer (${h2_error_code_name(frame.error_code)})', retryable)
                c.wake_send(mut s)
            }
        }
        H2ContinuationFrame {
            // CONTINUATION outside a header block is a connection error; the
            // in-block ones are consumed by on_response_headers.
            return error('unexpected CONTINUATION frame')
        }
        H2PushPromiseFrame {
            // We advertise SETTINGS_ENABLE_PUSH=0, so any PUSH_PROMISE is a
            // connection error (RFC 7540 6.6 / 8.2). Failing the connection
            // here also avoids the dropped fragment desyncing our HPACK decoder.
            return error('h2: unexpected PUSH_PROMISE (server push is disabled)')
        }
        else {
            // PRIORITY / unknown frame types: ignored per RFC 7540.
        }
    }
}

// apply_peer_settings folds the peer's SETTINGS into the connection,
// including the retroactive initial-window delta for every open stream
// (RFC 7540 6.9.2). Out-of-range values that would corrupt our framing or
// flow-control math are rejected as a connection error (the caller turns the
// error into fail_conn), per RFC 7540 6.5.2 / 6.5.3.
fn (mut c H2MuxConn) apply_peer_settings(settings []H2Setting) ! {
    for st in settings {
        match st.id {
            h2_settings_header_table_size {
                // RFC 7541 §6.3: even if our encoder uses only literals, we MUST
                // emit a Dynamic Table Size Update prefix at the start of the next
                // HEADERS block when the peer lowers this limit. encode() emits
                // the update when pending_max_table_size >= 0.
                c.wmu.lock()
                c.encoder.dyn_table.set_max_size(int(st.value))
                c.encoder.pending_max_table_size = int(st.value)
                c.wmu.unlock()
            }
            h2_settings_enable_push {
                if st.value > 1 {
                    // RFC 7540 6.5.2: ENABLE_PUSH must be 0 or 1.
                    return error('h2: peer SETTINGS_ENABLE_PUSH ${st.value} is not 0 or 1')
                }
                // We never use server push, so the value is otherwise irrelevant.
            }
            h2_settings_max_concurrent_streams {
                c.smu.lock()
                c.peer_max_streams = st.value
                c.smu.unlock()
            }
            h2_settings_max_header_list_size {
                // RFC 9113 §6.5.2: advisory cap on the size of the header list we
                // send. Store it (under wmu, with the other send-side header state)
                // so do_on_stream can refuse an over-limit request locally instead
                // of having the server reject it after the round trip.
                c.wmu.lock()
                c.peer_max_header_list_size = st.value
                c.wmu.unlock()
            }
            h2_settings_initial_window_size {
                if st.value > u32(0x7fff_ffff) {
                    // RFC 7540 6.5.3: values above 2^31-1 are a FLOW_CONTROL_ERROR;
                    // they also overflow our i64 send-window arithmetic.
                    return error('h2: peer SETTINGS_INITIAL_WINDOW_SIZE ${st.value} exceeds 2^31-1')
                }
                // smu is held across the fmu section so the snapshot of open
                // streams and the delta application are atomic with respect to
                // stream registration (which nests the same way) — a stream can
                // neither miss the delta nor receive it twice.
                c.smu.lock()
                c.fmu.lock()
                delta := i64(st.value) - c.peer_initial_window
                c.peer_initial_window = i64(st.value)
                // RFC 7540 §6.9.2: validate ALL stream windows before applying any
                // delta so that an overflow on stream N does not leave streams 1..N-1
                // in a partially updated state.  Pre-validation also ensures the
                // broadcast below is always reached for a positive delta.
                for _, s in c.streams {
                    if s.send_window + delta > i64(0x7fff_ffff) {
                        c.fmu.unlock()
                        c.smu.unlock()
                        return error('h2: SETTINGS_INITIAL_WINDOW_SIZE delta overflows stream ${s.id} send window (RFC 7540 §6.9.2 FLOW_CONTROL_ERROR)')
                    }
                }
                for _, mut s in c.streams {
                    s.send_window += delta
                }
                if delta > 0 {
                    c.fcv.broadcast()
                }
                c.fmu.unlock()
                c.smu.unlock()
            }
            h2_settings_max_frame_size {
                if st.value < h2_default_max_frame_size || st.value > u32(0x00ff_ffff) {
                    // RFC 7540 6.5.2: valid range is 2^14..2^24-1. A value such as
                    // 0 would make our HEADERS/DATA chunk step 0 and hang the send
                    // path in a zero-length-frame loop, so fail the connection.
                    return error('h2: peer SETTINGS_MAX_FRAME_SIZE ${st.value} out of range [16384, 16777215]')
                }
                // Take wmu (then fmu, the permitted wmu -> fmu nesting) before
                // lowering the cap. The send paths re-read peer_max_frame under fmu
                // and then write the frame under wmu, holding wmu across both; an
                // fmu-only update could land in the fmu-release -> write gap and let
                // a sender emit a frame larger than the peer's new limit
                // (FRAME_SIZE_ERROR). Serializing on wmu makes the senders' re-read
                // authoritative, mirroring the header_table_size arm above.
                // apply_peer_settings holds no connection lock at entry, so this
                // wmu acquisition cannot invert any held lock.
                c.wmu.lock()
                c.fmu.lock()
                c.peer_max_frame = st.value
                c.fmu.unlock()
                c.wmu.unlock()
            }
            else {} // unknown settings are ignored (RFC 7540 6.5.2)
        }
    }
}

// on_response_headers assembles a complete header block (reading any
// CONTINUATION frames inline — the reader owns the read path), decodes it,
// and delivers it to the stream. Blocks for unknown streams are still decoded
// to keep the connection's HPACK dynamic table in sync, then dropped.
fn (mut c H2MuxConn) on_response_headers(frame H2HeadersFrame) ! {
    mut fragment := frame.fragment.clone()
    if !frame.end_headers {
        for {
            cont := c.mux_read_frame() or {
                return error('connection lost inside a header block: ${err.msg()}')
            }
            if cont is H2ContinuationFrame {
                if cont.stream_id != frame.stream_id {
                    return error('CONTINUATION on the wrong stream')
                }
                fragment << cont.fragment
                if fragment.len > h2_max_recv_header_block {
                    return error('h2: response header block exceeds ${h2_max_recv_header_block} bytes')
                }
                if cont.end_headers {
                    break
                }
            } else {
                return error('expected a CONTINUATION frame')
            }
        }
    }
    fields := c.decoder.decode(fragment)!
    mut s := c.lookup_stream(frame.stream_id)
    if s == unsafe { nil } {
        return c.check_unknown_stream(frame.stream_id)
    }
    s.mu.lock()
    // RFC 9113 §5.1: a HEADERS frame after the stream has already ended (a prior
    // END_STREAM, or RST/connection death that set s.ended) is a frame on a closed
    // stream, not more trailers. Without this, back-to-back trailer + extra HEADERS
    // in one read buffer would be appended to resp.headers before wait_response
    // flushes, delivering fields the peer sent after closing the stream. Reset
    // instead (the DATA path makes the same check via s.ended).
    if s.ended {
        s.mu.unlock()
        c.reset_stream(frame.stream_id, .stream_closed,
            'HEADERS frame after stream end on stream ${frame.stream_id}')
        return
    }
    was_headers_done := s.headers_done
    if !s.headers_done {
        mut status := 0
        mut status_valid := false
        for f in fields {
            if f.name == ':status' {
                // RFC 9113 §8.3.1: :status is exactly three digits. string.int()
                // is lenient ('200 OK' -> 200), so validate the raw value before
                // converting; otherwise a malformed status is accepted as valid.
                if f.value.len == 3 && all_digits(f.value) {
                    status = f.value.int()
                    status_valid = true
                }
                break
            }
        }
        // A response MUST carry a valid :status (RFC 9113 §8.3.1), and HTTP/2
        // forbids 101 (§8.1.1). A missing, malformed, out-of-range, or 101 status
        // is a stream-level PROTOCOL_ERROR — reset it rather than treat it as a
        // 1xx interim and wait forever for a "final" HEADERS that never arrives.
        // (Trailers legitimately omit :status, but they are handled below since
        // headers_done is already set by then.)
        if !status_valid || status < 100 || status > 599 || status == 101 {
            s.mu.unlock()
            c.reset_stream(frame.stream_id, .protocol_error,
                'response with a missing or invalid :status')
            return
        }
        // RFC 9110 §15.2 / RFC 9113 §8.1: a server may send 1xx interim responses
        // (100 Continue, 103 Early Hints) before the final response. They are not
        // the final response and carry no body, so ignore them and keep waiting
        // for the final (>= 200) HEADERS rather than latching the interim status
        // and headers — which would make the real final HEADERS look like trailers.
        if status >= 200 {
            s.status = status
            mut seen_regular := false
            mut seen_status := false
            for f in fields {
                if f.name.starts_with(':') {
                    // RFC 9113 §8.3: the only valid response pseudo-header is :status
                    // (consumed above); pseudo-headers MUST precede regular fields and
                    // MUST NOT be duplicated. Any other ':' field, :status after a
                    // regular field, or a second :status makes the response malformed.
                    if f.name != ':status' || seen_regular || seen_status {
                        s.mu.unlock()
                        c.reset_stream(frame.stream_id, .protocol_error,
                            'malformed response: invalid pseudo-header ${f.name}')
                        return
                    }
                    seen_status = true
                    continue
                }
                seen_regular = true
                // RFC 9113 §8.2: reject malformed field names (uppercase, empty) and
                // connection-specific fields rather than delivering them to the caller.
                reason := h2_response_field_error(f.name)
                if reason != '' {
                    s.mu.unlock()
                    c.reset_stream(frame.stream_id, .protocol_error,
                        'malformed response: ${reason}')
                    return
                }
                // RFC 9110 §8.6 / RFC 9113 §8.1.1: a malformed Content-Length makes
                // the message malformed (a stream-level PROTOCOL_ERROR). u64() is
                // lenient ('12junk' -> 12, '0x10' -> 16), so validate it strictly
                // here before wait_response trusts it for the body-length check.
                if f.name == 'content-length' && !all_digits(f.value) {
                    s.mu.unlock()
                    c.reset_stream(frame.stream_id, .protocol_error,
                        'malformed Content-Length in response')
                    return
                }
                s.resp_headers << f
            }
            s.headers_done = true
        }
    }
    // Trailers without END_STREAM violate RFC 7540 §8.1 (trailers must carry
    // END_STREAM); reset the stream rather than hanging wait_response forever.
    if was_headers_done && !frame.end_stream {
        s.mu.unlock()
        c.reset_stream(frame.stream_id, .protocol_error,
            'trailers HEADERS frame must carry END_STREAM')
        return
    }
    // A 1xx informational HEADERS with END_STREAM is also forbidden (RFC 9113 §8.1).
    if !s.headers_done && frame.end_stream {
        s.mu.unlock()
        c.reset_stream(frame.stream_id, .protocol_error, '1xx response must not carry END_STREAM')
        return
    }
    // A second HEADERS block on the stream carries trailers (RFC 9113 §8.1).
    // Preserve its non-pseudo fields (grpc-status, digest, ...) so callers on
    // the mux path keep the trailer metadata the synchronous
    // H2Conn.read_response surfaces. Pseudo-header fields are forbidden in
    // trailers (§8.1) and any malformed field name (§8.2) makes the message
    // malformed, so both reset the stream. wait_response flushes the kept fields
    // into resp.headers when the stream ends. (The trailers-without-END_STREAM
    // case was already reset above, so a trailer block here always carries
    // END_STREAM.)
    if was_headers_done {
        for f in fields {
            // RFC 9113 §8.1: trailers MUST NOT contain pseudo-header fields, and the
            // §8.2 field-name rules apply as for any header block.
            if f.name.starts_with(':') {
                s.mu.unlock()
                c.reset_stream(frame.stream_id, .protocol_error,
                    'malformed trailers: pseudo-header ${f.name}')
                return
            }
            reason := h2_response_field_error(f.name)
            if reason != '' {
                s.mu.unlock()
                c.reset_stream(frame.stream_id, .protocol_error, 'malformed trailers: ${reason}')
                return
            }
            s.resp_trailers << f
        }
    }
    if frame.end_stream {
        s.ended = true
    }
    s.cv.signal()
    s.mu.unlock()
    if frame.end_stream {
        // The server closed the stream: wake any body sender that is blocked
        // waiting for flow-control credit, so it does not hang forever when
        // the peer withholds WINDOW_UPDATEs after an early final response.
        c.wake_send(mut s)
    }
}

// on_response_data delivers a DATA payload to its stream, or — for recently
// cancelled/completed streams — keeps the connection-level flow-control
// account exact by returning the credit directly (mirroring the server's
// unknown-stream backstop).
fn (mut c H2MuxConn) on_response_data(frame H2DataFrame) ! {
    // Padding (the pad-length byte + padding) counts toward flow control
    // (RFC 7540 6.9.1) but is never delivered to the app, so credit it back
    // immediately; the data bytes are credited as the requester drains them.
    pad_overhead := frame.flow_size - frame.data.len
    // Enforce the connection-level receive window (RFC 7540 §6.9). Debit
    // first; credit paths (drain, unknown-stream, cancelled) restore it.
    // recv_wmu is always acquired solo, so no lock-order hazard here.
    c.recv_wmu.lock()
    c.conn_recv_window -= i64(frame.flow_size)
    conn_ok := c.conn_recv_window >= 0
    c.recv_wmu.unlock()
    if !conn_ok {
        return error('h2: peer exceeded connection-level receive window (RFC 7540 §6.9 FLOW_CONTROL_ERROR)')
    }
    mut s := c.lookup_stream(frame.stream_id)
    if s == unsafe { nil } {
        c.check_unknown_stream(frame.stream_id)!
        // Stream is gone: credit the whole received payload to the connection.
        if frame.flow_size > 0 {
            c.send_conn_window_update(u32(frame.flow_size)) or {}
        }
        return
    }
    s.mu.lock()
    if s.cancelled {
        // The requester abandoned this stream (it has been deregistered, but the
        // reader still held a reference). Account for the whole payload at the
        // connection level and drop it, so the window does not leak. Checking
        // the flag under s.mu — the same lock cancel_stream sets it under —
        // makes this race-free against a frame in flight during cancellation.
        s.mu.unlock()
        if frame.flow_size > 0 {
            c.send_conn_window_update(u32(frame.flow_size)) or {}
        }
        return
    }
    if !s.headers_done {
        // RFC 7540 §8.1: an HTTP/2 response must begin with HEADERS. DATA before
        // the final header block is malformed — RST this stream (PROTOCOL_ERROR)
        // rather than queue bytes a requester would see with status 0. Credit the
        // frame back to the connection window (debited above, never delivered).
        s.mu.unlock()
        if frame.flow_size > 0 {
            c.send_conn_window_update(u32(frame.flow_size)) or {}
        }
        c.reset_stream(frame.stream_id, .protocol_error,
            'DATA before response HEADERS on stream ${frame.stream_id}')
        return
    }
    // RFC 7540 §5.1: a DATA frame after the peer's END_STREAM is a stream-closed
    // error. If we have also closed our send side the stream is "closed" — a
    // connection error (the spec's MUST); otherwise it is "half-closed (remote)"
    // — a STREAM error, so one peer's misbehavior does not tear down every other
    // multiplexed stream. Checked before the flow-control debit: a half-closed
    // (remote) receiver is no longer obligated to maintain the window (§6.9.1).
    if s.ended {
        send_closed := s.send_closed
        s.mu.unlock()
        if send_closed {
            return error('h2: DATA frame after END_STREAM on closed stream ${frame.stream_id} (RFC 7540 §5.1)')
        }
        if frame.flow_size > 0 {
            c.send_conn_window_update(u32(frame.flow_size)) or {}
        }
        c.reset_stream(frame.stream_id, .stream_closed,
            'DATA after END_STREAM on stream ${frame.stream_id}')
        return
    }
    // Enforce the stream-level receive window (RFC 7540 §6.9.1). Padding
    // counts against stream flow control the same as data, so debit
    // whenever flow_size > 0 regardless of whether there are data bytes.
    if frame.flow_size > 0 {
        s.recv_window -= i64(frame.flow_size)
        if s.recv_window < 0 {
            s.mu.unlock()
            // RFC 7540 §6.9.1: a stream-level flow-control violation is a STREAM
            // error — RST just this stream (like the WINDOW_UPDATE overflow path)
            // rather than failing the whole connection and every other multiplexed
            // stream. Credit this frame's bytes back to the connection window
            // (debited above, never delivered); reset_stream credits queued chunks.
            c.send_conn_window_update(u32(frame.flow_size)) or {}
            c.reset_stream(frame.stream_id, .flow_control_error,
                'peer exceeded stream ${frame.stream_id} receive window')
            return
        }
    }
    if frame.data.len > 0 {
        s.chunks << frame.data.clone()
        s.body_rcvd += u64(frame.data.len)
    }
    if frame.end_stream {
        s.ended = true
    }
    s.cv.signal()
    s.mu.unlock()
    if pad_overhead > 0 {
        c.send_window_updates(s.id, u32(pad_overhead)) or {}
    }
    if frame.end_stream {
        // The response ended on this DATA frame (an early final response with a
        // body, e.g. a 413/auth rejection). Wake any body sender still blocked on
        // flow-control credit, mirroring on_response_headers; otherwise it hangs
        // when the peer withholds further WINDOW_UPDATEs.
        c.wake_send(mut s)
    }
}

// check_unknown_stream distinguishes late frames for cancelled/finished
// streams (fine) from protocol garbage (connection error).
fn (mut c H2MuxConn) check_unknown_stream(stream_id u32) ! {
    c.wmu.lock()
    known_range := stream_id < c.next_stream_id
    c.wmu.unlock()
    if stream_id % 2 == 1 && known_range {
        return
    }
    return error('frame for an unknown stream ${stream_id}')
}

// wake_send marks a stream's send side dead and wakes a body sender that is
// blocked waiting for flow-control credit, so a RST_STREAM or GOAWAY terminates
// the upload instead of hanging it. The caller must hold no connection lock
// (it takes fmu). fail() must already have recorded the terminal error.
fn (mut c H2MuxConn) wake_send(mut s H2MuxStream) {
    c.fmu.lock()
    s.send_dead = true
    c.fcv.broadcast()
    c.fmu.unlock()
}

// reset_stream deregisters a stream, sends RST_STREAM(code), and fails its
// requester — used by the reader for a stream-level protocol/flow-control error.
// Deregistering before the RST lets any in-flight late DATA hit the
// unknown-stream backstop (which credits the connection window).
fn (mut c H2MuxConn) reset_stream(stream_id u32, code H2ErrorCode, reason string) {
    c.smu.lock()
    mut s := c.streams[stream_id] or { &H2MuxStream(unsafe { nil }) }
    c.streams.delete(stream_id)
    c.smu.unlock()
    c.wmu.lock()
    c.write_all_locked(H2Frame(H2RstStreamFrame{
        stream_id:  stream_id
        error_code: u32(code)
    }).encode()) or {
        c.wmu.unlock()
        c.note_write_failure()
        // The stream was removed from c.streams above, so fail_conn() will not
        // find it. Wake it directly so the requester does not block forever.
        if s != unsafe { nil } {
            s.fail('h2: transport write failure', false)
            c.wake_send(mut s)
        }
        return
    }
    c.wmu.unlock()
    if s != unsafe { nil } {
        // Credit back any DATA bytes already queued but never drained,
        // mirroring cancel_stream, so the connection receive window stays
        // accurate after a per-stream reset.
        s.mu.lock()
        mut queued := u64(0)
        for ch in s.chunks {
            queued += u64(ch.len)
        }
        s.chunks.clear()
        s.mu.unlock()
        if queued > 0 {
            c.send_conn_window_update(u32(queued)) or {}
        }
        s.fail('h2: ${reason}', false)
        c.wake_send(mut s)
    }
}

fn (mut c H2MuxConn) lookup_stream(stream_id u32) &H2MuxStream {
    c.smu.lock()
    defer {
        c.smu.unlock()
    }
    if s := c.streams[stream_id] {
        return s
    }
    return &H2MuxStream(unsafe { nil })
}

// fail_conn marks the connection dead, fails every in-flight stream, and wakes
// all blocked senders. Normally called by the reader on transport error; also
// called by note_write_failure when close_transport is nil and cannot interrupt
// the reader. Guards against double-invocation via the c.closed check under smu.
// Teardown is routed through teardown_transport, which runs close_transport
// exactly once, so this path and drop_ref's last-reference teardown can never
// double-close.
fn (mut c H2MuxConn) fail_conn(msg string) {
    c.smu.lock()
    if c.closed {
        c.smu.unlock()
        return
    }
    c.closed = true
    c.conn_err = msg
    mut open := []&H2MuxStream{}
    for _, s in c.streams {
        open << s
    }
    c.streams.clear()
    pending_teardown := c.refs <= 0
    c.smu.unlock()
    for mut s in open {
        s.mu.lock()
        retryable := !s.sent_headers
        s.mu.unlock()
        s.fail(msg, retryable)
    }
    c.fmu.lock()
    c.fmu_dead = true
    c.fcv.broadcast()
    c.fmu.unlock()
    if pending_teardown {
        c.teardown_transport()
    }
}