Crafter.Network/implementations/Crafter.Network-ClientQUIC.cpp

//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

module;
#include <msquic.h>
#include <cstring>
module Crafter.Network:ClientQUIC_impl;
import :ClientQUIC;
import Crafter.Thread;
import std;

using namespace Crafter;

namespace {
    // Process-wide msquic API table + registration. Initialised lazily on
    // first ClientQUIC/ListenerQUIC construction; tear-down happens at
    // process exit via the destructor of the static object.
    struct MsQuicRuntime {
        const QUIC_API_TABLE* api = nullptr;
        HQUIC registration = nullptr;
        std::mutex initMutex;
        bool initialised = false;

        void Ensure() {
            std::lock_guard lock(initMutex);
            if (initialised) return;
            QUIC_STATUS s = MsQuicOpen2(&api);
            if (QUIC_FAILED(s)) {
                throw QUICException(std::format("MsQuicOpen2 failed: 0x{:x}", static_cast<unsigned>(s)));
            }
            QUIC_REGISTRATION_CONFIG regConfig{ "crafter.network", QUIC_EXECUTION_PROFILE_LOW_LATENCY };
            s = api->RegistrationOpen(&regConfig, &registration);
            if (QUIC_FAILED(s)) {
                MsQuicClose(api);
                api = nullptr;
                throw QUICException(std::format("RegistrationOpen failed: 0x{:x}", static_cast<unsigned>(s)));
            }
            initialised = true;
        }

        ~MsQuicRuntime() {
            if (registration) api->RegistrationClose(registration);
            if (api) MsQuicClose(api);
        }
    };

    MsQuicRuntime& Runtime() {
        static MsQuicRuntime r;
        r.Ensure();
        return r;
    }

    // Open-stream bookkeeping, shared by a connection and every stream on it.
    // A stream counts as open from the moment msquic has a callback handler
    // for it until its handle has been StreamClose'd. ~ClientQUIC drains this
    // before closing the connection: QUICStream::Stop only *initiates* a
    // shutdown whose async completion does the actual close, so without the
    // wait a connection can be closed with streams still open on the msquic
    // side.
    //
    // Reached through a shared_ptr rather than through the owning ClientQUIC*
    // so that a stream finalising after the (bounded) wait gave up decrements
    // a live object instead of a freed connection.
    struct StreamRegistry {
        std::mutex mtx;
        std::condition_variable cv;
        int open = 0;

        void Add() {
            std::lock_guard lk(mtx);
            ++open;
        }

        void Remove() {
            {
                std::lock_guard lk(mtx);
                --open;
            }
            cv.notify_all();
        }

        // False if the timeout expired with streams still open.
        bool WaitForDrain(std::chrono::milliseconds timeout) {
            std::unique_lock lk(mtx);
            return cv.wait_for(lk, timeout, [&]{ return open == 0; });
        }
    };

    // ClientQUIC::OnClosed's state, shared by the ClientQUIC and whoever
    // fires the callback: the SHUTDOWN_COMPLETE handler on an msquic worker,
    // or a pool task when OnClosed is called after the close. Reached
    // through a shared_ptr because the handler fires it after closing the
    // connection handle, and from then on ~ClientQUIC, finding no handle
    // left to claim, no longer waits for the worker and may free the Impl
    // under it.
    //
    // The rules are std::stop_callback's: once ~ClientQUIC has set
    // `ownerGone` the callback never starts; one already running on another
    // thread is waited for, so the owner may free what the callback uses as
    // soon as the destructor returns; one running on the destructor's own
    // thread, a callback deleting its own connection, cannot be and is not.
    struct CloseNotice {
        std::mutex mtx;
        std::condition_variable idle;
        std::function<void()> callback;
        bool closed = false;      // shut down, and not by ~ClientQUIC
        bool fired = false;
        bool ownerGone = false;   // ~ClientQUIC has begun
        bool running = false;
        std::thread::id runningOn;

        // Runs the callback if it is due: closed, set, not yet run, and the
        // owner still there.
        static void Fire(const std::shared_ptr<CloseNotice>& notice) {
            std::function<void()> callback;
            {
                std::lock_guard lk(notice->mtx);
                if (!notice->closed || notice->fired || notice->ownerGone || !notice->callback) return;
                notice->fired = true;
                notice->running = true;
                notice->runningOn = std::this_thread::get_id();
                callback.swap(notice->callback);
            }
            callback();
            // Destroyed before `running` drops, so a destructor waiting for
            // us is never racing what the callback captured.
            callback = nullptr;
            {
                std::lock_guard lk(notice->mtx);
                notice->running = false;
            }
            notice->idle.notify_all();
        }

        // ~ClientQUIC's half: from here no callback starts, and one running
        // on another thread has returned.
        void OwnerGone() {
            std::function<void()> dropped;
            {
                std::unique_lock lk(mtx);
                ownerGone = true;
                if (running && runningOn != std::this_thread::get_id()) {
                    idle.wait(lk, [&] { return !running; });
                }
                dropped.swap(callback);
            }
        }
    };

    // Bound on how long ~ClientQUIC waits for the connection's streams to
    // finish closing. The decrements arrive on msquic worker threads, so an
    // unbounded wait would turn a dropped peer into a hung destructor.
    //
    // Expiring is a bug, not a slow path: ConnectionShutdown makes msquic
    // deliver SHUTDOWN_COMPLETE for every stream it knows about, OpenStream
    // passes SHUTDOWN_ON_FAIL so a stream whose start lost the race to that
    // shutdown gets one too, and QUICStream::Stop aborts the receive direction
    // so a bidi stream does not sit waiting on a peer that will never close
    // its send side. With those three the count always reaches zero; the bound
    // only stops an msquic-side surprise from wedging the destructor outright.
    constexpr std::chrono::milliseconds streamDrainTimeout{ 5000 };

    // Bound on how long ~ClientQUIC then waits for the connection's shutdown
    // to complete before it closes the handle. Closing earlier loses the
    // CONNECTION_CLOSE frame: msquic sends it from a flush it queues while it
    // processes ConnectionShutdown, a ConnectionClose that reaches the worker
    // before that flush completes the shutdown silently (QuicConnTryClose's
    // "silent close forced" path), and the flush is then skipped (FLUSH_SEND
    // is ignored once ShutdownComplete is set). The peer learned of the close
    // only at its idle timeout, two minutes by default; with no stream open
    // the drain above returns at once, so that was the usual outcome.
    // SHUTDOWN_COMPLETE comes about a round trip later, once the peer answers
    // the close, or after msquic's closing period (three PTOs) when it does
    // not. Either way the frame went out at the start, so the bound only
    // stops a peer that is gone from holding the destructor that long.
    constexpr std::chrono::milliseconds closeNotifyTimeout{ 1000 };

    // Encode an ALPN string into the wire format msquic expects: a length
    // byte followed by the ASCII characters. Lifetime of the returned buffer
    // matches the caller's storage in `out`.
    QUIC_BUFFER MakeAlpn(const std::string& alpn, std::vector<std::uint8_t>& out) {
        if (alpn.size() > 255) throw QUICException("ALPN string too long (max 255)");
        out.assign(alpn.begin(), alpn.end());
        QUIC_BUFFER b{};
        b.Length = static_cast<std::uint32_t>(out.size());
        b.Buffer = out.data();
        return b;
    }
}

// ---------------- QUICStream::Impl ----------------
struct QUICStream::Impl {
    // How one SendSync went, written by its SEND_COMPLETE under the stream's
    // mutex. Shared with the waiting thread so the request can be freed by
    // whichever side finishes with it, in either order.
    struct SendOutcome {
        bool done = false;
        bool sent = false;
    };

    // One send handed to msquic: the copy it reads from, and the QUIC_BUFFER
    // describing it, which msquic keeps a pointer to rather than copying.
    // Owned by msquic from a successful StreamSend until that send's
    // SEND_COMPLETE, which msquic delivers exactly once for every send it
    // took (Canceled when the stream or the connection ended first) and
    // always before the stream's SHUTDOWN_COMPLETE; that handler frees it.
    // Who hears how it went: SendSync's waiting thread through `outcome`,
    // or SendAsync's `onSent`.
    struct SendRequest {
        std::vector<char> bytes;
        QUIC_BUFFER buffer{};
        std::shared_ptr<SendOutcome> outcome;
        std::function<void(bool sent)> onSent;

        SendRequest(const void* data, std::uint32_t size)
            : bytes(static_cast<const char*>(data), static_cast<const char*>(data) + size) {
            buffer.Buffer = reinterpret_cast<std::uint8_t*>(bytes.data());
            buffer.Length = size;
        }
    };

    ClientQUIC* connection = nullptr;

    std::mutex mtx;
    std::condition_variable cv;
    std::deque<std::vector<char>> pending;
    bool peerSendClosed = false;
    bool shutdownComplete = false;

    // Sends handed to msquic whose SEND_COMPLETE has not come yet, under mtx.
    std::uint64_t bytesInFlight = 0;
    std::size_t sendsInFlight = 0;

    // StreamSend for one request, counted in flight while msquic has it.
    // False, with the request still the caller's, when msquic refused it:
    // then no SEND_COMPLETE will come.
    bool Send(HQUIC live, SendRequest* request, bool finish, QUIC_STATUS& status) {
        {
            std::lock_guard lk(mtx);
            bytesInFlight += request->bytes.size();
            ++sendsInFlight;
        }
        // Counted first: SEND_COMPLETE can run on the worker before
        // StreamSend has returned here.
        status = Runtime().api->StreamSend(live, &request->buffer, 1,
                                           finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE, request);
        if (QUIC_SUCCEEDED(status)) return true;
        std::lock_guard lk(mtx);
        bytesInFlight -= request->bytes.size();
        --sendsInFlight;
        return false;
    }

    // ---- Callbacks: OnRecieve's onData and onEnd, SendAsync's onSent ----
    // They run one at a time per stream, in order, and never under `mtx`.
    // Whichever thread finds nobody delivering becomes the delivering thread
    // and runs what is ready, picking up whatever others queue meanwhile; the
    // others only queue and return. Normally that thread is the msquic
    // worker, inside the event that made something ready, so no thread is
    // held while nothing is; it is the caller's thread for the bytes
    // OnRecieve finds waiting, and a pool task for a send that never reached
    // msquic. Nothing blocks to wait for the delivering thread, the worker
    // least of all, except Unregister, as std::stop_callback does.
    //
    // A blocking alternative, one mutex held across every callback, would
    // make the worker wait whenever the caller's thread was delivering, and a
    // callback that called into msquic and waited for that worker
    // (GetStreamId does) would deadlock; a pool task per delivery would make
    // every byte depend on a pool the host shares.
    std::function<void(std::span<const char>)> onData;
    std::function<void(bool)> onEnd;
    bool handlerSet = false;      // OnRecieve was called
    bool unregistered = false;    // Stop(): onData and onEnd never again
    bool aborted = false;         // Abort(): no more onData, what waits is dropped
    enum class End { Open, Clean, Unclean };
    End end = End::Open;          // how the receive direction ended, once it has
    bool endDelivered = false;
    bool delivering = false;
    std::thread::id deliveringThread;
    std::condition_variable deliveryIdle;
    std::deque<std::pair<std::function<void(bool)>, bool>> completions;   // onSent calls waiting their turn

    // Becomes the delivering thread if there is none. Under mtx.
    bool BeginDelivery() {
        if (delivering) return false;
        delivering = true;
        deliveringThread = std::this_thread::get_id();
        return true;
    }

    // Runs what is ready, one callback at a time and none under the lock,
    // until nothing is, then stops delivering. Entered as the delivering
    // thread with `lk` held; returns with it released.
    void DeliverReady(std::unique_lock<std::mutex>& lk) {
        for (;;) {
            if (!completions.empty()) {
                std::function<void(bool)> onSent = std::move(completions.front().first);
                const bool sent = completions.front().second;
                completions.pop_front();
                lk.unlock();
                onSent(sent);
                onSent = nullptr;
                lk.lock();
                continue;
            }
            if (!handlerSet || unregistered) break;
            if (aborted) pending.clear();
            if (!pending.empty()) {
                std::vector<char> chunk = std::move(pending.front());
                pending.pop_front();
                lk.unlock();
                onData(std::span<const char>(chunk.data(), chunk.size()));
                lk.lock();
                continue;
            }
            // Only once everything before it has been handed over.
            if (end != End::Open && !endDelivered) {
                endDelivered = true;
                const bool clean = end == End::Clean;
                lk.unlock();
                onEnd(clean);
                lk.lock();
                continue;
            }
            break;
        }
        // Nothing calls onData or onEnd again once the end is out or the
        // owner unregistered, so let them go, with what they captured: a
        // callback holding its own stream would otherwise keep it forever.
        // Destroyed outside the lock, where a capture's destructor may do
        // anything.
        std::function<void(std::span<const char>)> dataHandler;
        std::function<void(bool)> endHandler;
        if (handlerSet && (endDelivered || unregistered)) {
            dataHandler.swap(onData);
            endHandler.swap(onEnd);
        }
        delivering = false;
        lk.unlock();
        deliveryIdle.notify_all();
    }

    // Delivers what is ready if no thread is, and releases `lk` either way.
    void DeliverIfIdle(std::unique_lock<std::mutex>& lk) {
        if (BeginDelivery()) {
            DeliverReady(lk);
        } else {
            lk.unlock();
        }
    }

    // Queues onSent(sent) behind the stream's other callbacks.
    void Complete(std::function<void(bool)> onSent, bool sent, std::unique_lock<std::mutex>& lk) {
        completions.emplace_back(std::move(onSent), sent);
        DeliverIfIdle(lk);
    }

    // A SendAsync that never reached msquic still reports, exactly once, but
    // never from inside SendAsync, which would deadlock a caller holding a
    // lock its own callback takes: after the callback that made it, when one
    // is being delivered, else from a pool task.
    static void FailSend(const std::shared_ptr<Impl>& self, std::function<void(bool sent)> onSent) {
        if (!onSent) return;
        if (self) {
            std::lock_guard lk(self->mtx);
            if (self->delivering) {
                self->completions.emplace_back(std::move(onSent), false);
                return;
            }
        }
        ThreadPool::Enqueue([self, onSent = std::move(onSent)]() mutable {
            if (!self) {
                onSent(false);
                return;
            }
            std::unique_lock lk(self->mtx);
            self->Complete(std::move(onSent), false, lk);
        });
    }

    // Stop()'s half: onData and onEnd never run once this returns, and one
    // running on another thread has finished. One running on this thread is
    // the caller, a callback destroying its own wrapper; it cannot be waited
    // for, and the delivering loop lets the handlers go when it unwinds.
    void Unregister() {
        std::function<void(std::span<const char>)> dataHandler;
        std::function<void(bool)> endHandler;
        std::unique_lock lk(mtx);
        unregistered = true;
        if (!handlerSet) return;
        if (delivering && deliveringThread != std::this_thread::get_id()) {
            deliveryIdle.wait(lk, [&] { return !delivering; });
        }
        if (!delivering) {
            dataHandler.swap(onData);
            endHandler.swap(onEnd);
        }
        lk.unlock();
    }

    // Up to `count` received bytes, waiting for them; fewer once the peer
    // has closed its side and the rest is taken.
    std::vector<char> TakeUpTo(std::uint32_t count) {
        std::vector<char> out;
        out.reserve(count);
        std::unique_lock lk(mtx);
        while (out.size() < count) {
            cv.wait(lk, [&]{ return !pending.empty() || peerSendClosed || shutdownComplete; });
            Take(out, count);
            if (pending.empty() && (peerSendClosed || shutdownComplete)) break;
        }
        return out;
    }

    // Exactly `count` received bytes, once they have all arrived; nothing,
    // leaving what did arrive where it is, once the peer has closed its side
    // short of them. One hold of the lock throughout, so no other receive
    // takes bytes from among them.
    std::optional<std::vector<char>> TakeExactly(std::uint32_t count) {
        std::unique_lock lk(mtx);
        auto arrived = [&] {
            std::size_t bytes = 0;
            for (const std::vector<char>& chunk : pending) bytes += chunk.size();
            return bytes;
        };
        cv.wait(lk, [&]{ return arrived() >= count || peerSendClosed || shutdownComplete; });
        if (arrived() < count) return std::nullopt;
        std::vector<char> out;
        out.reserve(count);
        Take(out, count);
        return out;
    }

    // Moves pending bytes onto `out` until it holds `count`. Under mtx.
    void Take(std::vector<char>& out, std::size_t count) {
        while (!pending.empty() && out.size() < count) {
            auto& chunk = pending.front();
            std::size_t want = std::min<std::size_t>(chunk.size(), count - out.size());
            out.insert(out.end(), chunk.begin(), chunk.begin() + want);
            if (want == chunk.size()) {
                pending.pop_front();
            } else {
                chunk.erase(chunk.begin(), chunk.begin() + want);
            }
        }
    }

    // Lifetime: msquic delivers stream callbacks (RECEIVE, SHUTDOWN_COMPLETE, …)
    // possibly AFTER the owning QUICStream wrapper is destroyed (the wrapper only
    // initiates a graceful shutdown; completion is async). To make that safe, the
    // callback context is a heap `shared_ptr<Impl>*` (selfRef) that holds one
    // strong ref; every callback copies it locally so the Impl stays alive for
    // the callback's duration. The terminal SHUTDOWN_COMPLETE callback closes the
    // stream and releases selfRef, after which the Impl is freed once the last
    // ref (wrapper / in-flight callback copy) drops.
    std::shared_ptr<Impl>* selfRef = nullptr;

    // The owning connection's open-stream count. Incremented where the msquic
    // handler is installed (QUICStream's handle constructor / OpenStream),
    // decremented once the handle has been closed.
    std::shared_ptr<StreamRegistry> registry;

    // The connection's registry. Defined below ClientQUIC::Impl (which owns
    // it); a member of QUICStream::Impl rather than a free function so it
    // inherits QUICStream's friendship with ClientQUIC.
    static std::shared_ptr<StreamRegistry> RegistryOf(ClientQUIC* connection);

    // ---- msquic handle ownership ----------------------------------------
    // StreamClose is "equivalent to free": any other msquic call on the handle
    // that overlaps it is a use-after-free, which msquic reports as a bugcheck
    // on whichever thread trips over the wreckage. The handle therefore has a
    // refcount, and whoever drops the last ref closes it. There are two kinds
    // of holder:
    //
    //   - The msquic callback holds exactly one ref for the stream's lifetime,
    //     released by Retire() once SHUTDOWN_COMPLETE has been delivered (or by
    //     the OpenStream error paths, where no callback will ever run).
    //   - Every app thread inside an msquic call on the handle holds one for
    //     the duration of that call (AcquireHandle / ReleaseHandle).
    //
    // Without the second kind, a thread sending on a stream races the worker
    // thread that closes it when the connection is torn down underneath —
    // the shape any client with periodic acks or keepalives has.
    std::mutex handleMtx;
    HQUIC handle = nullptr;    // cleared when the last ref lets go
    int handleRefs = 1;        // starts with the callback's ref
    bool retired = false;      // the callback's ref has been released

    // The handle, valid until the matching ReleaseHandle; nullptr once the
    // stream is retired, in which case do not call ReleaseHandle.
    HQUIC AcquireHandle() {
        std::lock_guard lk(handleMtx);
        if (retired || !handle) return nullptr;
        ++handleRefs;
        return handle;
    }

    void ReleaseHandle() {
        HQUIC toClose = nullptr;
        {
            std::lock_guard lk(handleMtx);
            if (--handleRefs > 0) return;
            toClose = handle;
            handle = nullptr;
        }
        // Last ref: nobody else can be inside an msquic call on this handle,
        // and AcquireHandle will hand it out no more.
        if (toClose) Runtime().api->StreamClose(toClose);
        std::shared_ptr<StreamRegistry> reg = std::move(registry);
        std::shared_ptr<Impl>* ref = selfRef;
        selfRef = nullptr;
        delete ref;   // may destroy *this — touch no members past here
        // Last of all: a destructor blocked in WaitForDrain may run
        // ConnectionClose the moment this hits zero, and msquic wants every
        // StreamClose on the connection to have happened first.
        if (reg) reg->Remove();
    }

    // Publish the msquic handle. Called once, while the stream is still local
    // to its constructor and no callback can run on it.
    void SetHandle(HQUIC h) {
        std::lock_guard lk(handleMtx);
        handle = h;
    }

    // Release the callback's ref. Idempotent, so the SHUTDOWN_COMPLETE handler
    // and an OpenStream error path can both call it without racing.
    void Retire() {
        {
            std::lock_guard lk(handleMtx);
            if (retired) return;
            retired = true;
        }
        ReleaseHandle();
    }

    static QUIC_STATUS QUIC_API Callback(HQUIC stream, void* ctx, QUIC_STREAM_EVENT* ev) {
        // Keep the Impl alive for the whole callback, even if the wrapper (and
        // its ref) go away concurrently.
        std::shared_ptr<Impl> keepAlive = *static_cast<std::shared_ptr<Impl>*>(ctx);
        Impl* self = keepAlive.get();
        switch (ev->Type) {
            case QUIC_STREAM_EVENT_RECEIVE: {
                // Everything is consumed here (SUCCESS), whoever ends up with
                // it: a pending receive would hold the stream's shutdown back
                // until a StreamReceiveComplete that nothing would ever call.
                std::unique_lock lk(self->mtx);
                // What Abort() has not stopped yet is dropped, and so is
                // whatever comes once nothing will hand it over.
                if (self->aborted
                    || (self->handlerSet && (self->unregistered || self->endDelivered))) {
                    return QUIC_STATUS_SUCCESS;
                }
                if (!self->handlerSet || self->delivering || !self->pending.empty()) {
                    // For a Recieve* call, or behind what another thread is
                    // still handing over: a copy, in order.
                    std::vector<char> chunk;
                    std::uint64_t total = 0;
                    for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) {
                        total += ev->RECEIVE.Buffers[i].Length;
                    }
                    chunk.reserve(static_cast<std::size_t>(total));
                    for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) {
                        const QUIC_BUFFER& b = ev->RECEIVE.Buffers[i];
                        chunk.insert(chunk.end(), b.Buffer, b.Buffer + b.Length);
                    }
                    if (!chunk.empty()) self->pending.push_back(std::move(chunk));
                    if (!self->handlerSet) {
                        lk.unlock();
                        self->cv.notify_all();
                        return QUIC_STATUS_SUCCESS;
                    }
                    self->DeliverIfIdle(lk);
                    return QUIC_STATUS_SUCCESS;
                }
                // Nothing waiting and nobody delivering: hand msquic's own
                // buffers over, valid until this event returns, without a copy.
                self->BeginDelivery();
                lk.unlock();
                for (std::uint32_t i = 0; i < ev->RECEIVE.BufferCount; ++i) {
                    const QUIC_BUFFER& b = ev->RECEIVE.Buffers[i];
                    if (b.Length == 0) continue;
                    if (i > 0) {
                        // The last onData may have aborted or dropped the stream.
                        std::lock_guard check(self->mtx);
                        if (self->aborted || self->unregistered) break;
                    }
                    self->onData(std::span<const char>(reinterpret_cast<const char*>(b.Buffer), b.Length));
                }
                lk.lock();
                self->DeliverReady(lk);
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_SEND_COMPLETE: {
                auto* request = static_cast<SendRequest*>(ev->SEND_COMPLETE.ClientContext);
                if (request == nullptr) return QUIC_STATUS_SUCCESS;
                const bool sent = !ev->SEND_COMPLETE.Canceled;
                {
                    std::lock_guard lk(self->mtx);
                    self->bytesInFlight -= request->bytes.size();
                    --self->sendsInFlight;
                    if (request->outcome) {
                        request->outcome->done = true;
                        request->outcome->sent = sent;
                    }
                }
                if (request->outcome) self->cv.notify_all();
                std::function<void(bool)> onSent = std::move(request->onSent);
                delete request;
                // SendAsync's, on the worker, in turn with the stream's other
                // callbacks: msquic is done with the copy, and no thread was
                // held while it had it.
                if (onSent) {
                    std::unique_lock lk(self->mtx);
                    self->Complete(std::move(onSent), sent, lk);
                }
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN:
            case QUIC_STREAM_EVENT_PEER_SEND_ABORTED: {
                // PEER_SEND_SHUTDOWN comes once every byte up to the FIN has
                // been taken from RECEIVE events, so onEnd(true) follows the
                // last onData.
                {
                    std::unique_lock lk(self->mtx);
                    self->peerSendClosed = true;
                    if (self->end == End::Open) {
                        self->end = ev->Type == QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN ? End::Clean : End::Unclean;
                    }
                    self->DeliverIfIdle(lk);
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: {
                {
                    std::unique_lock lk(self->mtx);
                    self->peerSendClosed = true;
                    // Every send msquic took has had its SEND_COMPLETE by now;
                    // a SendSync still waiting is only woken to find out.
                    self->shutdownComplete = true;
                    // Neither a FIN nor the peer's reset ended the receive
                    // side: the connection went, or this side aborted it.
                    if (self->end == End::Open) self->end = End::Unclean;
                    // Before Retire, so the handle is still valid for the
                    // callbacks, as in every other event.
                    self->DeliverIfIdle(lk);
                }
                self->cv.notify_all();
                self->Retire();
                return QUIC_STATUS_SUCCESS;
            }
            default:
                return QUIC_STATUS_SUCCESS;
        }
    }
};

QUICStream::QUICStream() = default;

QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection)
    : handle(handle), connection(connection), impl(std::make_shared<Impl>())
{
    impl->SetHandle(handle);   // handleRefs already accounts for the callback
    impl->connection = connection;
    impl->registry = Impl::RegistryOf(connection);
    if (impl->registry) impl->registry->Add();
    impl->selfRef = new std::shared_ptr<Impl>(impl);   // strong ref owned by the callback
    Runtime().api->SetCallbackHandler(handle, reinterpret_cast<void*>(&Impl::Callback), impl->selfRef);
}

QUICStream::QUICStream(QUICStream&& other) noexcept
    : handle(other.handle), connection(other.connection),
      canSend(other.canSend), canReceive(other.canReceive),
      impl(std::move(other.impl))
{
    other.handle = nullptr;
    other.connection = nullptr;
}

QUICStream& QUICStream::operator=(QUICStream&& other) noexcept {
    if (this != &other) {
        Stop();
        handle = other.handle;
        connection = other.connection;
        canSend = other.canSend;
        canReceive = other.canReceive;
        impl = std::move(other.impl);
        other.handle = nullptr;
        other.connection = nullptr;
    }
    return *this;
}

QUICStream::~QUICStream() {
    Stop();
}

void QUICStream::Stop() {
    if (!handle) return;
    handle = nullptr;   // the wrapper is done with the stream either way
    if (!impl) return;
    // The owner is done with OnRecieve's callbacks too.
    impl->Unregister();
    // Only INITIATE the shutdown here; the async SHUTDOWN_COMPLETE callback
    // releases the handle, and the Impl is a shared_ptr held by the callback's
    // selfRef, so it safely outlives this wrapper until that callback runs.
    // A null handle means SHUTDOWN_COMPLETE already fired and msquic has closed
    // the stream — shutting it down again would be a use-after-free.
    HQUIC live = impl->AcquireHandle();
    if (!live) return;
    // GRACEFUL closes our send direction only, and msquic rejects it combined
    // with any other flag, so the receive direction needs a call of its own.
    // Without it a bidirectional stream whose peer keeps its send side open
    // never reaches SHUTDOWN_COMPLETE, so its handle is never closed and the
    // connection can never finish tearing down.
    Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0);
    if (canReceive) {
        Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, 0);
    }
    impl->ReleaseHandle();
}

void QUICStream::SendSync(const void* buffer, std::uint32_t size, bool finish) {
    if (!handle || !canSend) throw QUICClosedException();
    // Hold the handle across StreamSend: without this a connection torn down
    // on another thread can close the stream between the check above and the
    // call below, and msquic bugchecks on the freed handle.
    HQUIC live = impl->AcquireHandle();
    if (!live) throw QUICClosedException();
    // This send's own outcome, not the stream's state. The flag this used to
    // wait on was one per stream, and the wait then threw whenever the
    // stream had finished shutting down by the time it woke: a peer that
    // answered and finished the stream within a round trip, quicker than a
    // busy machine wakes this thread, made a send that went out throw.
    auto outcome = std::make_shared<Impl::SendOutcome>();
    auto* request = new Impl::SendRequest(buffer, size);
    request->outcome = outcome;
    QUIC_STATUS s = QUIC_STATUS_SUCCESS;
    const bool taken = impl->Send(live, request, finish, s);
    impl->ReleaseHandle();
    if (!taken) {
        // No SEND_COMPLETE will arrive for a send msquic never took.
        delete request;
        throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    std::unique_lock lk(impl->mtx);
    // SEND_COMPLETE always precedes the stream's SHUTDOWN_COMPLETE; waking on
    // the latter only keeps an msquic surprise from hanging this thread.
    impl->cv.wait(lk, [&]{ return outcome->done || impl->shutdownComplete; });
    if (!outcome->done || !outcome->sent) throw QUICClosedException();
}

std::vector<char> QUICStream::RecieveSync() {
    if (!handle || !canReceive) throw QUICClosedException();
    std::unique_lock lk(impl->mtx);
    impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; });
    if (!impl->pending.empty()) {
        auto out = std::move(impl->pending.front());
        impl->pending.pop_front();
        return out;
    }
    throw QUICClosedException();
}

std::vector<char> QUICStream::RecieveUntilCloseSync() {
    if (!handle || !canReceive) throw QUICClosedException();
    std::vector<char> out;
    while (true) {
        std::unique_lock lk(impl->mtx);
        impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; });
        while (!impl->pending.empty()) {
            auto& chunk = impl->pending.front();
            out.insert(out.end(), chunk.begin(), chunk.end());
            impl->pending.pop_front();
        }
        if (impl->peerSendClosed || impl->shutdownComplete) return out;
    }
}

std::vector<char> QUICStream::RecieveUntilFullSync(std::uint32_t bufferSize) {
    if (!handle || !canReceive) throw QUICClosedException();
    // Closed early: what did arrive stays for the next receive.
    std::optional<std::vector<char>> out = impl->TakeExactly(bufferSize);
    if (!out) throw QUICClosedException();
    return std::move(*out);
}

void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish,
                           std::function<void(bool sent)> onSent) {
    // This used to be SendSync run as a pool task: a pool thread held for as
    // long as msquic took to accept the buffer (a peer that reads slowly
    // stretches that to its pace), and a task that reached back through
    // `this`, so moving or destroying the wrapper before it ran was a
    // use-after-free. Now the send goes to msquic here and its SEND_COMPLETE
    // reports, on the worker. Only `impl` is read, not `handle`, which Stop()
    // clears: SendAsync may run on another thread than the owner's, inside a
    // callback, and a stopped stream refuses or cancels the send anyway.
    HQUIC live = impl && canSend ? impl->AcquireHandle() : nullptr;
    if (live == nullptr) {
        Impl::FailSend(impl, std::move(onSent));
        return;
    }
    auto* request = new Impl::SendRequest(buffer, size);
    request->onSent = std::move(onSent);
    QUIC_STATUS status = QUIC_STATUS_SUCCESS;
    const bool taken = impl->Send(live, request, finish, status);
    impl->ReleaseHandle();
    if (!taken) {
        std::function<void(bool)> callback = std::move(request->onSent);
        delete request;
        Impl::FailSend(impl, std::move(callback));
    }
}

std::uint64_t QUICStream::BytesInFlight() const {
    if (!impl) return 0;
    std::lock_guard lk(impl->mtx);
    return impl->bytesInFlight;
}

std::size_t QUICStream::SendsInFlight() const {
    if (!impl) return 0;
    std::lock_guard lk(impl->mtx);
    return impl->sendsInFlight;
}

// The async receives deliver a closed stream as the browser build does: an
// empty chunk, as nothing may leave a pool task.
void QUICStream::RecieveAsync(std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, cb]{
        std::vector<char> chunk;
        try {
            chunk = this->RecieveSync();
        } catch (const QUICClosedException&) {}
        cb(std::move(chunk));
    });
}
void QUICStream::RecieveUntilCloseAsync(std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, cb]{
        std::vector<char> data;
        try {
            data = this->RecieveUntilCloseSync();
        } catch (const QUICClosedException&) {}
        cb(std::move(data));
    });
}
void QUICStream::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function<void(std::vector<char>)> cb) {
    ThreadPool::Enqueue([this, bufferSize, cb]{
        cb(handle && canReceive ? impl->TakeUpTo(bufferSize) : std::vector<char>{});
    });
}

void QUICStream::PrependReceived(std::vector<char> bytes) {
    if (bytes.empty() || !impl) return;
    std::unique_lock lk(impl->mtx);
    impl->pending.push_front(std::move(bytes));
    if (impl->handlerSet && !impl->unregistered) {
        // Handed to onData next, by this thread or the one delivering.
        impl->DeliverIfIdle(lk);
        return;
    }
    lk.unlock();
    impl->cv.notify_all();
}

void QUICStream::OnRecieve(std::function<void(std::span<const char> data)> onData,
                           std::function<void(bool clean)> onEnd) {
    // Our own reference: the callbacks delivered below may destroy this
    // wrapper, and nothing of `this` is touched once they can run.
    const std::shared_ptr<Impl> self = impl;
    if (!self || !handle) {
        if (onEnd) onEnd(false);
        return;
    }
    std::unique_lock lk(self->mtx);
    if (self->handlerSet) {
        throw QUICException("OnRecieve: this stream's callbacks are already set");
    }
    self->handlerSet = true;
    self->onData = onData ? std::move(onData) : [](std::span<const char>) {};
    self->onEnd = onEnd ? std::move(onEnd) : [](bool) {};
    // A stream only we send on gets no data and no FIN to end cleanly with.
    if (!canReceive && self->end == Impl::End::Open) self->end = Impl::End::Unclean;
    // What arrived before now, and the end if it came: delivered here, unless
    // another thread is delivering this stream's callbacks (a send's onSent),
    // which then hands these over after its own.
    self->DeliverIfIdle(lk);
}

void QUICStream::Abort(std::uint64_t errorCode) {
    const std::shared_ptr<Impl> self = impl;
    if (!self) return;
    {
        std::lock_guard lk(self->mtx);
        self->aborted = true;
        self->pending.clear();
        if (self->end == Impl::End::Open) self->end = Impl::End::Unclean;
    }
    // Inside a callback, the delivering loop hands onEnd(false) over when the
    // callback returns; otherwise SHUTDOWN_COMPLETE does, soon: IMMEDIATE
    // makes msquic report it as soon as it processes the abort, instead of
    // once the peer has acknowledged the reset, which a peer that is gone
    // never does.
    HQUIC live = self->AcquireHandle();
    if (live == nullptr) return;
    Runtime().api->StreamShutdown(live,
                                  QUIC_STREAM_SHUTDOWN_FLAG_ABORT_SEND
                                      | QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE
                                      | QUIC_STREAM_SHUTDOWN_FLAG_IMMEDIATE,
                                  std::min<std::uint64_t>(errorCode, QUIC_UINT62_MAX));
    self->ReleaseHandle();
}

std::uint64_t QUICStream::GetStreamId() const {
    if (!handle) throw QUICException("GetStreamId: stream is not open");
    // GetParam blocks on the connection's msquic worker, so it must not be the
    // worker that waits for us — hence a refcount rather than an exclusion
    // lock: SHUTDOWN_COMPLETE just drops its ref and returns.
    HQUIC live = impl->AcquireHandle();
    if (!live) throw QUICException("GetStreamId: stream is not open");
    QUIC_UINT62 id = 0;
    std::uint32_t size = sizeof(id);
    QUIC_STATUS s = Runtime().api->GetParam(live, QUIC_PARAM_STREAM_ID, &size, &id);
    impl->ReleaseHandle();
    if (QUIC_FAILED(s)) {
        throw QUICException(std::format("GetParam(QUIC_PARAM_STREAM_ID) failed: 0x{:x}",
                                        static_cast<unsigned>(s)));
    }
    return static_cast<std::uint64_t>(id);
}

// ---------------- ClientQUIC::Impl ----------------
struct ClientQUIC::Impl {
    HQUIC connection = nullptr;
    HQUIC configuration = nullptr;
    bool ownsConfiguration = true;

    std::mutex mtx;
    std::condition_variable cv;
    bool connected = false;
    bool closed = false;
    bool shutdownComplete = false;   // msquic delivered SHUTDOWN_COMPLETE
    QUIC_STATUS shutdownStatus = QUIC_STATUS_SUCCESS;

    std::function<void(QUICStream)> onStream;
    std::function<void(std::vector<char>)> onDatagram;
    std::deque<std::vector<char>> datagramQueue;
    // Streams the peer started before the user installed an OnStream
    // handler. Without this backlog the early streams (e.g. an h3 server's
    // control stream right after handshake) would be aborted in the
    // PEER_STREAM_STARTED branch and the connection would die with
    // H3_MISSING_SETTINGS on the peer side.
    std::deque<QUICStream> pendingStreams;

    // Streams currently open on this connection. Shared with each QUICStream
    // so the count survives a stream that outlives us.
    std::shared_ptr<StreamRegistry> streams = std::make_shared<StreamRegistry>();

    // OnClosed's callback and the state that decides whether it runs.
    std::shared_ptr<CloseNotice> closeNotice = std::make_shared<CloseNotice>();

    ClientQUIC* outer = nullptr;

    // Take the connection handle away from whoever else might close it.
    // `connection` is the single owner token: exactly one of ~ClientQUIC and
    // the SHUTDOWN_COMPLETE callback wins the claim, and only the winner calls
    // ConnectionClose. Without this the two race and msquic bugchecks on the
    // second close (CXPLAT_TEL_ASSERT(!Connection->State.HandleClosed)).
    HQUIC ClaimConnection() {
        std::lock_guard lk(mtx);
        HQUIC claimed = connection;
        connection = nullptr;
        return claimed;
    }

    // Shut the connection down and close it, waiting (bounded) for every
    // stream on it to reach StreamClose first, and then for the shutdown to
    // complete, so the peer is told. No-op if someone else already claimed
    // the handle. Never holds `mtx` across an msquic call: the connection
    // callback takes it too.
    void CloseConnection() {
        HQUIC claimed = ClaimConnection();
        if (!claimed) return;
        // Aborts any stream still open; each one's SHUTDOWN_COMPLETE runs
        // Retire, which closes it and drops the count below.
        Runtime().api->ConnectionShutdown(claimed, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
        streams->WaitForDrain(streamDrainTimeout);
        {
            // See closeNotifyTimeout. The SHUTDOWN_COMPLETE handler sets the
            // flag before it looks at the handle, which we hold, so it leaves
            // the close to us.
            std::unique_lock lk(mtx);
            cv.wait_for(lk, closeNotifyTimeout, [&] { return shutdownComplete; });
        }
        Runtime().api->ConnectionClose(claimed);
    }

    static QUIC_STATUS QUIC_API Callback(HQUIC conn, void* ctx, QUIC_CONNECTION_EVENT* ev) {
        auto* self = static_cast<Impl*>(ctx);
        switch (ev->Type) {
            case QUIC_CONNECTION_EVENT_CONNECTED: {
                {
                    std::lock_guard lk(self->mtx);
                    self->connected = true;
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT:
            case QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_PEER: {
                {
                    std::lock_guard lk(self->mtx);
                    self->closed = true;
                    if (ev->Type == QUIC_CONNECTION_EVENT_SHUTDOWN_INITIATED_BY_TRANSPORT) {
                        self->shutdownStatus = ev->SHUTDOWN_INITIATED_BY_TRANSPORT.Status;
                    }
                }
                self->cv.notify_all();
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE: {
                {
                    std::lock_guard lk(self->mtx);
                    self->closed = true;
                    self->shutdownComplete = true;
                }
                self->cv.notify_all();
                // AppCloseInProgress means ~ClientQUIC is already inside
                // ConnectionClose; otherwise close the handle here so a
                // connection dropped by the peer does not leak while the
                // owning ClientQUIC lives on. Claiming the handle settles which
                // of the two actually does it. Every stream msquic knew of was
                // aborted at the first close (QuicConnTryClose), so each has
                // had its own SHUTDOWN_COMPLETE before this one.
                if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress != 0) return QUIC_STATUS_SUCCESS;
                std::shared_ptr<CloseNotice> notice;
                {
                    // The claim and the notice together, so ~ClientQUIC
                    // cannot slip between them.
                    std::lock_guard lk(self->mtx);
                    if (self->connection == nullptr) return QUIC_STATUS_SUCCESS;
                    self->connection = nullptr;
                    notice = self->closeNotice;
                }
                Runtime().api->ConnectionClose(conn);
                // Nothing of `self` from here on: ~ClientQUIC, finding no
                // handle to claim, does not wait for this worker, and the
                // OnClosed callback may well be what deletes it.
                {
                    std::lock_guard lk(notice->mtx);
                    notice->closed = true;
                }
                CloseNotice::Fire(notice);
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_PEER_STREAM_STARTED: {
                HQUIC streamHandle = ev->PEER_STREAM_STARTED.Stream;
                bool unidirectional = (ev->PEER_STREAM_STARTED.Flags
                                       & QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL) != 0;
                QUICStream stream(streamHandle, self->outer);
                if (unidirectional) {
                    // Peer-initiated unidi: peer sends, we read; we cannot send.
                    stream.canSend = false;
                    stream.canReceive = true;
                }
                std::function<void(QUICStream)> cb;
                {
                    std::lock_guard lk(self->mtx);
                    cb = self->onStream;
                    if (!cb) {
                        // Buffer until OnStream is installed; OnStream's
                        // setter drains this queue.
                        self->pendingStreams.push_back(std::move(stream));
                        return QUIC_STATUS_SUCCESS;
                    }
                }
                auto* shared = new QUICStream(std::move(stream));
                ThreadPool::Enqueue([cb, shared]{
                    cb(std::move(*shared));
                    delete shared;
                });
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_DATAGRAM_RECEIVED: {
                std::vector<char> chunk(ev->DATAGRAM_RECEIVED.Buffer->Buffer,
                                       ev->DATAGRAM_RECEIVED.Buffer->Buffer + ev->DATAGRAM_RECEIVED.Buffer->Length);
                if (self->onDatagram) {
                    auto cb = self->onDatagram;
                    ThreadPool::Enqueue([cb, chunk = std::move(chunk)]() mutable { cb(std::move(chunk)); });
                } else {
                    std::lock_guard lk(self->mtx);
                    self->datagramQueue.push_back(std::move(chunk));
                    self->cv.notify_all();
                }
                return QUIC_STATUS_SUCCESS;
            }
            case QUIC_CONNECTION_EVENT_DATAGRAM_SEND_STATE_CHANGED: {
                // msquic fires this event multiple times per datagram (e.g.
                // SENT -> ACKNOWLEDGED). Free the combined QUIC_BUFFER+payload
                // allocation only on a terminal state — SENT and LOST_SUSPECT
                // are intermediate and may be followed by another transition.
                auto state = ev->DATAGRAM_SEND_STATE_CHANGED.State;
                if (ev->DATAGRAM_SEND_STATE_CHANGED.ClientContext &&
                    (state == QUIC_DATAGRAM_SEND_LOST_DISCARDED
                     || state == QUIC_DATAGRAM_SEND_ACKNOWLEDGED
                     || state == QUIC_DATAGRAM_SEND_ACKNOWLEDGED_SPURIOUS
                     || state == QUIC_DATAGRAM_SEND_CANCELED)) {
                    ::operator delete(ev->DATAGRAM_SEND_STATE_CHANGED.ClientContext);
                }
                return QUIC_STATUS_SUCCESS;
            }
            default:
                return QUIC_STATUS_SUCCESS;
        }
    }
};

static HQUIC OpenClientConfiguration(const std::string& alpn, const QUICClientCredentials& creds) {
    std::vector<std::uint8_t> alpnBuf;
    QUIC_BUFFER alpnBuffer = MakeAlpn(alpn, alpnBuf);

    QUIC_SETTINGS settings{};
    settings.IsSet.IdleTimeoutMs = 1;
    settings.IdleTimeoutMs = creds.settings.idleTimeoutMs;
    settings.IsSet.HandshakeIdleTimeoutMs = 1;
    settings.HandshakeIdleTimeoutMs = creds.settings.handshakeIdleTimeoutMs;
    // Keep the connection alive across long idle gaps. msquic sends PING frames
    // on its own timer thread (independent of the app), so a request/response
    // connection survives even while the app is blocked for a long time between
    // requests (e.g. a client waiting on slow LLM inference). Interval < idle
    // timeout so the peer's idle timer never expires.
    settings.IsSet.KeepAliveIntervalMs = 1;
    settings.KeepAliveIntervalMs = creds.settings.keepAliveIntervalMs;
    settings.IsSet.DatagramReceiveEnabled = 1;
    settings.DatagramReceiveEnabled = creds.settings.datagramReceiveEnabled ? 1 : 0;
    // Allow the server to open unidi/bidi streams to us. msquic defaults
    // both peer-stream-count limits to 0; with that, the server's HTTP/3
    // control stream + QPACK encoder/decoder streams can't be created and
    // most h3 servers will close the connection after handshake. We don't
    // currently use server push (h3 pushes ride on unidi 0x01 streams) but
    // the bidi cap is harmless to grant.
    settings.IsSet.PeerUnidiStreamCount = 1;
    settings.PeerUnidiStreamCount = creds.settings.peerUnidiStreamCount;
    settings.IsSet.PeerBidiStreamCount = 1;
    settings.PeerBidiStreamCount = creds.settings.peerBidiStreamCount;

    HQUIC cfg = nullptr;
    QUIC_STATUS s = Runtime().api->ConfigurationOpen(Runtime().registration, &alpnBuffer, 1,
                                                     &settings, sizeof(settings), nullptr, &cfg);
    if (QUIC_FAILED(s)) throw QUICException(std::format("ConfigurationOpen failed: 0x{:x}", static_cast<unsigned>(s)));

    QUIC_CREDENTIAL_CONFIG cc{};
    cc.Type = QUIC_CREDENTIAL_TYPE_NONE;
    cc.Flags = QUIC_CREDENTIAL_FLAG_CLIENT;
    if (creds.insecureNoServerValidation) {
        cc.Flags |= QUIC_CREDENTIAL_FLAG_NO_CERTIFICATE_VALIDATION;
    }
    s = Runtime().api->ConfigurationLoadCredential(cfg, &cc);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConfigurationClose(cfg);
        throw QUICException(std::format("ConfigurationLoadCredential failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    return cfg;
}

ClientQUIC::ClientQUIC(const char* host, std::uint16_t port, std::string alpnIn,
                       QUICClientCredentials creds)
    : alpn(std::move(alpnIn)), impl(std::make_unique<Impl>())
{
    impl->outer = this;
    impl->configuration = OpenClientConfiguration(alpn, creds);

    QUIC_STATUS s = Runtime().api->ConnectionOpen(Runtime().registration,
                                                  reinterpret_cast<QUIC_CONNECTION_CALLBACK_HANDLER>(&Impl::Callback),
                                                  impl.get(), &impl->connection);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConfigurationClose(impl->configuration);
        throw QUICException(std::format("ConnectionOpen failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    s = Runtime().api->ConnectionStart(impl->connection, impl->configuration, QUIC_ADDRESS_FAMILY_UNSPEC, host, port);
    if (QUIC_FAILED(s)) {
        Runtime().api->ConnectionClose(impl->connection);
        Runtime().api->ConfigurationClose(impl->configuration);
        throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast<unsigned>(s)));
    }

    QUIC_STATUS handshakeStatus = QUIC_STATUS_SUCCESS;
    {
        std::unique_lock lk(impl->mtx);
        // Bounded by HandshakeIdleTimeoutMs above: an unreachable peer ends in
        // SHUTDOWN_INITIATED_BY_TRANSPORT rather than waiting here forever.
        impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; });
        if (impl->connected) return;
        handshakeStatus = impl->shutdownStatus;
    }
    // Throwing from here means ~ClientQUIC never runs, so tear the msquic
    // handles down by hand — the connection callback holds `impl`, which is
    // about to be destroyed with the half-built object.
    impl->CloseConnection();
    Runtime().api->ConfigurationClose(impl->configuration);
    impl->configuration = nullptr;
    throw QUICException(std::format("QUIC handshake failed: 0x{:x}",
                                    static_cast<unsigned>(handshakeStatus)));
}

ClientQUIC::ClientQUIC(std::string host, std::uint16_t port, std::string alpnIn, QUICClientCredentials creds)
    : ClientQUIC(host.c_str(), port, std::move(alpnIn), std::move(creds)) {}

ClientQUIC::ClientQUIC(HQUIC connectionHandle, HQUIC serverConfiguration, std::string alpnIn)
    : alpn(std::move(alpnIn)), impl(std::make_unique<Impl>())
{
    impl->outer = this;
    impl->connection = connectionHandle;
    impl->configuration = serverConfiguration;
    impl->ownsConfiguration = false;
    impl->connected = true;
    Runtime().api->SetCallbackHandler(connectionHandle,
                                      reinterpret_cast<void*>(&Impl::Callback),
                                      impl.get());
}

ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept
    : alpn(std::move(other.alpn)), impl(std::move(other.impl))
{
    if (impl) impl->outer = this;
}

ClientQUIC::~ClientQUIC() {
    if (!impl) return;
    // First of all: the OnClosed callback must not start once the owner has
    // begun destroying the connection, and one running on another thread
    // has to be done with it before it goes. A callback that is deleting
    // this very object runs on this thread and is not waited for; it already
    // closed the handle, so CloseConnection below finds nothing to claim and
    // returns at once.
    impl->closeNotice->OwnerGone();
    // Before the connection goes: streams the peer started that no OnStream
    // handler ever claimed. Destroying them here rather than with the rest of
    // `impl` keeps every stream teardown ahead of ConnectionClose, so the
    // drain below actually covers them.
    {
        std::deque<QUICStream> backlog;
        {
            std::lock_guard lk(impl->mtx);
            std::swap(backlog, impl->pendingStreams);
        }
    }
    impl->CloseConnection();
    if (impl->configuration && impl->ownsConfiguration) {
        Runtime().api->ConfigurationClose(impl->configuration);
        impl->configuration = nullptr;
    }
}

void ClientQUIC::Stop() {
    if (!impl) return;
    // Read the handle under the lock: the SHUTDOWN_COMPLETE callback may be
    // closing the connection concurrently on an msquic worker.
    HQUIC conn = nullptr;
    {
        std::lock_guard lk(impl->mtx);
        conn = impl->connection;
    }
    if (conn) Runtime().api->ConnectionShutdown(conn, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
}

QUICStream ClientQUIC::OpenStream(bool unidirectional) {
    HQUIC streamHandle = nullptr;
    QUICStream stream;
    stream.impl = std::make_shared<QUICStream::Impl>();
    stream.impl->connection = this;
    // Count the stream as open before StreamOpen registers the callback, so a
    // callback can never observe a half-initialised Impl. The two failure
    // paths below undo it.
    stream.impl->registry = impl->streams;
    stream.impl->registry->Add();
    stream.impl->selfRef = new std::shared_ptr<QUICStream::Impl>(stream.impl);
    QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional
        ? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL
        : QUIC_STREAM_OPEN_FLAG_NONE;
    QUIC_STATUS s = Runtime().api->StreamOpen(impl->connection, openFlags,
                                              reinterpret_cast<QUIC_STREAM_CALLBACK_HANDLER>(&QUICStream::Impl::Callback),
                                              stream.impl->selfRef, &streamHandle);
    if (QUIC_FAILED(s)) {
        delete stream.impl->selfRef;
        stream.impl->selfRef = nullptr;
        // No msquic stream exists, so there is no handle to release.
        std::shared_ptr<StreamRegistry> registry = std::move(stream.impl->registry);
        registry->Remove();
        throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    stream.handle = streamHandle;
    stream.connection = this;
    stream.impl->SetHandle(streamHandle);
    if (unidirectional) {
        // We initiated the unidi stream: we send, peer reads.
        stream.canSend = true;
        stream.canReceive = false;
    }
    // SHUTDOWN_ON_FAIL, not NONE. StreamStart is asynchronous — it returns
    // QUIC_STATUS_PENDING and queues the start onto the connection — so it can
    // still fail later, and it does exactly that (QUIC_STATUS_INVALID_STATE)
    // whenever the connection is shut down before the queued start runs. That
    // is an everyday race for a client that sends periodically: opening a
    // stream just as the connection goes away. Without this flag msquic leaves
    // such a stream neither started nor shut down, so SHUTDOWN_COMPLETE never
    // arrives, the handle is never closed, and the registration's rundown at
    // exit() blocks forever waiting for it. With it, the failed start is
    // followed by SHUTDOWN_COMPLETE like any other stream.
    s = Runtime().api->StreamStart(streamHandle, QUIC_STREAM_START_FLAG_SHUTDOWN_ON_FAIL);
    if (QUIC_FAILED(s)) {
        // Synchronous failure: the stream never started, so no callback will
        // ever run. Release the callback's ref by hand, which closes the
        // handle, and clear the wrapper's so ~QUICStream is a no-op.
        stream.handle = nullptr;
        stream.impl->Retire();
        throw QUICException(std::format("StreamStart failed: 0x{:x}", static_cast<unsigned>(s)));
    }
    return stream;
}

void ClientQUIC::SendDatagram(const void* buffer, std::uint32_t size) {
    // msquic stores the QUIC_BUFFER pointer (not a copy) on the send queue
    // and serialises async on a worker thread. Both the QUIC_BUFFER and the
    // payload it points at must outlive the call until DATAGRAM_SEND_STATE
    // reports a terminal state. Pack them together in a single allocation.
    auto* mem = static_cast<std::uint8_t*>(::operator new(sizeof(QUIC_BUFFER) + size));
    auto* hdr = reinterpret_cast<QUIC_BUFFER*>(mem);
    auto* payload = mem + sizeof(QUIC_BUFFER);
    std::memcpy(payload, buffer, size);
    hdr->Buffer = payload;
    hdr->Length = size;
    QUIC_STATUS s = Runtime().api->DatagramSend(impl->connection, hdr, 1,
                                                QUIC_SEND_FLAG_NONE, mem);
    if (QUIC_FAILED(s)) {
        ::operator delete(mem);
        throw QUICException(std::format("DatagramSend failed: 0x{:x}", static_cast<unsigned>(s)));
    }
}

void ClientQUIC::OnStream(std::function<void(QUICStream)> cb) {
    std::deque<QUICStream> backlog;
    {
        std::lock_guard lk(impl->mtx);
        impl->onStream = cb;
        std::swap(backlog, impl->pendingStreams);
    }
    while (!backlog.empty()) {
        auto* shared = new QUICStream(std::move(backlog.front()));
        backlog.pop_front();
        auto handler = cb;
        ThreadPool::Enqueue([handler, shared]{
            handler(std::move(*shared));
            delete shared;
        });
    }
}

void ClientQUIC::OnDatagram(std::function<void(std::vector<char>)> cb) {
    impl->onDatagram = std::move(cb);
}

void ClientQUIC::OnClosed(std::function<void()> callback) {
    if (!impl) return;
    const std::shared_ptr<CloseNotice> notice = impl->closeNotice;
    bool due = false;
    {
        std::lock_guard lk(notice->mtx);
        if (notice->fired) return;
        // The one replaced is destroyed outside the lock, with `callback`.
        callback.swap(notice->callback);
        due = notice->closed && notice->callback;
    }
    // Closed already, typically before a ListenerQUIC's connectCallback got
    // to run: fire from the pool, never from in here, where the caller is
    // still using the object the callback is likely to delete. Fire checks
    // again, so a second OnClosed racing this one cannot fire it twice.
    if (due) ThreadPool::Enqueue([notice] { CloseNotice::Fire(notice); });
}

std::vector<char> ClientQUIC::RecieveDatagramSync() {
    std::unique_lock lk(impl->mtx);
    impl->cv.wait(lk, [&]{ return !impl->datagramQueue.empty() || impl->closed; });
    if (!impl->datagramQueue.empty()) {
        auto out = std::move(impl->datagramQueue.front());
        impl->datagramQueue.pop_front();
        return out;
    }
    throw QUICClosedException();
}

HQUIC ClientQUIC::GetHandle() const { return impl ? impl->connection : nullptr; }

std::shared_ptr<StreamRegistry> QUICStream::Impl::RegistryOf(ClientQUIC* connection) {
    if (!connection || !connection->impl) return nullptr;
    return connection->impl->streams;
}