ffc118f825
A client that sends on a stream while its connection is being torn down —
the shape any long-lived client with periodic acks or keepalives has —
would hang at exit inside MsQuicRegistrationClose, and occasionally
bugcheck on an msquic worker thread instead.
Root cause: StreamStart is asynchronous. It returns QUIC_STATUS_PENDING
and queues the start onto the connection, so it can still fail later, and
it fails with QUIC_STATUS_INVALID_STATE whenever the connection is shut
down before the queued start runs. Passed QUIC_STREAM_START_FLAG_NONE,
msquic leaves such a stream neither started nor shut down, so
SHUTDOWN_COMPLETE never arrives. The stream close is driven entirely off
that event, so the handle stays open: ~ClientQUIC's bounded drain expires,
the connection is closed with a stream msquic still considers open, and
the registration's rundown at exit() waits forever for the handle.
Instrumenting the handle lifecycle over the reproducer shows one leaked
locally-opened stream per failing run, correlating 1:1 with a
START_COMPLETE carrying 0x1 (INVALID_STATE).
Fixes:
- OpenStream passes QUIC_STREAM_START_FLAG_SHUTDOWN_ON_FAIL, so a start
that loses the race is followed by SHUTDOWN_COMPLETE like any other
stream and its handle is closed.
- QUICStream::Stop also aborts the receive direction. GRACEFUL closes
only the send side and msquic rejects it combined with any other flag,
so a bidirectional stream whose peer keeps its send side open never
reached SHUTDOWN_COMPLETE either.
- The msquic stream handle is now refcounted, held by the callback plus
every app thread inside an msquic call on it, and closed by whoever
lets go last. StreamSend/StreamShutdown/GetParam could previously run
against a handle a worker thread had just closed — a use-after-free
msquic reports as a bugcheck. A refcount rather than an exclusion lock
because GetParam blocks on the connection's worker, which must never
be the thread waiting.
- ~ClientQUIC drops undispatched peer streams before closing the
connection rather than with the rest of Impl afterwards, so the drain
covers them.
The reproducer (tests/ShouldSurviveConnectionChurn) grows a client-opened
bidirectional control stream with a thread sending across the teardown,
and no longer ends in std::_Exit — returning from main runs the static
MsQuicRuntime destructor, and its RegistrationClose is where a leaked
handle shows up. Before: 5 of 15 runs clean, 10 hung. After: 90 of 90
clean, with every opened handle observed closed and the drain never
expiring.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
887 lines
38 KiB
C++
887 lines
38 KiB
C++
//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(®Config, ®istration);
|
|
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; });
|
|
}
|
|
};
|
|
|
|
// 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 };
|
|
|
|
// 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 {
|
|
ClientQUIC* connection = nullptr;
|
|
|
|
std::mutex mtx;
|
|
std::condition_variable cv;
|
|
std::deque<std::vector<char>> pending;
|
|
bool peerSendClosed = false;
|
|
bool shutdownComplete = false;
|
|
bool sendInFlight = false;
|
|
|
|
// 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: {
|
|
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);
|
|
}
|
|
{
|
|
std::lock_guard lk(self->mtx);
|
|
if (!chunk.empty()) self->pending.push_back(std::move(chunk));
|
|
}
|
|
self->cv.notify_all();
|
|
return QUIC_STATUS_SUCCESS;
|
|
}
|
|
case QUIC_STREAM_EVENT_SEND_COMPLETE: {
|
|
{
|
|
std::lock_guard lk(self->mtx);
|
|
self->sendInFlight = false;
|
|
}
|
|
if (ev->SEND_COMPLETE.ClientContext) {
|
|
delete[] static_cast<char*>(ev->SEND_COMPLETE.ClientContext);
|
|
}
|
|
self->cv.notify_all();
|
|
return QUIC_STATUS_SUCCESS;
|
|
}
|
|
case QUIC_STREAM_EVENT_PEER_SEND_SHUTDOWN:
|
|
case QUIC_STREAM_EVENT_PEER_SEND_ABORTED: {
|
|
{
|
|
std::lock_guard lk(self->mtx);
|
|
self->peerSendClosed = true;
|
|
}
|
|
self->cv.notify_all();
|
|
return QUIC_STATUS_SUCCESS;
|
|
}
|
|
case QUIC_STREAM_EVENT_SHUTDOWN_COMPLETE: {
|
|
{
|
|
std::lock_guard lk(self->mtx);
|
|
self->peerSendClosed = true;
|
|
self->shutdownComplete = true;
|
|
// No further SEND_COMPLETE can arrive, so release anyone
|
|
// waiting on one — they throw on shutdownComplete instead.
|
|
self->sendInFlight = false;
|
|
}
|
|
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;
|
|
// 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();
|
|
auto* copy = new char[size];
|
|
std::memcpy(copy, buffer, size);
|
|
QUIC_BUFFER quicBuf{};
|
|
quicBuf.Buffer = reinterpret_cast<std::uint8_t*>(copy);
|
|
quicBuf.Length = size;
|
|
{
|
|
std::lock_guard lk(impl->mtx);
|
|
impl->sendInFlight = true;
|
|
}
|
|
QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE;
|
|
QUIC_STATUS s = Runtime().api->StreamSend(live, &quicBuf, 1, flags, copy);
|
|
impl->ReleaseHandle();
|
|
if (QUIC_FAILED(s)) {
|
|
{
|
|
// No SEND_COMPLETE will arrive for a send msquic never took.
|
|
std::lock_guard lk(impl->mtx);
|
|
impl->sendInFlight = false;
|
|
}
|
|
impl->cv.notify_all();
|
|
delete[] copy;
|
|
throw QUICException(std::format("StreamSend failed: 0x{:x}", static_cast<unsigned>(s)));
|
|
}
|
|
std::unique_lock lk(impl->mtx);
|
|
impl->cv.wait(lk, [&]{ return !impl->sendInFlight || impl->shutdownComplete; });
|
|
if (impl->shutdownComplete) 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();
|
|
std::vector<char> out;
|
|
out.reserve(bufferSize);
|
|
while (out.size() < bufferSize) {
|
|
std::unique_lock lk(impl->mtx);
|
|
impl->cv.wait(lk, [&]{ return !impl->pending.empty() || impl->peerSendClosed || impl->shutdownComplete; });
|
|
while (!impl->pending.empty() && out.size() < bufferSize) {
|
|
auto& chunk = impl->pending.front();
|
|
std::size_t want = std::min<std::size_t>(chunk.size(), bufferSize - out.size());
|
|
out.insert(out.end(), chunk.begin(), chunk.begin() + want);
|
|
if (want == chunk.size()) {
|
|
impl->pending.pop_front();
|
|
} else {
|
|
chunk.erase(chunk.begin(), chunk.begin() + want);
|
|
}
|
|
}
|
|
if (out.size() < bufferSize && (impl->peerSendClosed || impl->shutdownComplete)) {
|
|
throw QUICClosedException();
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
|
|
void QUICStream::SendAsync(const void* buffer, std::uint32_t size, bool finish,
|
|
std::function<void()> onSent) {
|
|
// Copy now: the caller's buffer may not outlive the enqueued task.
|
|
std::vector<char> copy(static_cast<const char*>(buffer),
|
|
static_cast<const char*>(buffer) + size);
|
|
ThreadPool::Enqueue([this, copy = std::move(copy), finish, onSent = std::move(onSent)]() mutable {
|
|
try { this->SendSync(copy.data(), static_cast<std::uint32_t>(copy.size()), finish); }
|
|
catch (...) { /* swallowed — callback still fires so the caller can move on */ }
|
|
if (onSent) onSent();
|
|
});
|
|
}
|
|
|
|
void QUICStream::RecieveAsync(std::function<void(std::vector<char>)> cb) {
|
|
ThreadPool::Enqueue([this, cb]{ cb(this->RecieveSync()); });
|
|
}
|
|
void QUICStream::RecieveUntilCloseAsync(std::function<void(std::vector<char>)> cb) {
|
|
ThreadPool::Enqueue([this, cb]{ cb(this->RecieveUntilCloseSync()); });
|
|
}
|
|
void QUICStream::RecieveUntilFullAsync(std::uint32_t bufferSize, std::function<void(std::vector<char>)> cb) {
|
|
ThreadPool::Enqueue([this, bufferSize, cb]{ cb(this->RecieveUntilFullSync(bufferSize)); });
|
|
}
|
|
|
|
void QUICStream::PrependReceived(std::vector<char> bytes) {
|
|
if (bytes.empty() || !impl) return;
|
|
{
|
|
std::lock_guard lk(impl->mtx);
|
|
impl->pending.push_front(std::move(bytes));
|
|
}
|
|
impl->cv.notify_all();
|
|
}
|
|
|
|
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;
|
|
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>();
|
|
|
|
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. 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);
|
|
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->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. ClaimConnection settles which
|
|
// of the two actually does it. Every stream on the connection
|
|
// has already been shut down (and so closed) by the time
|
|
// msquic reports the connection complete.
|
|
if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0
|
|
&& self->ClaimConnection() != nullptr) {
|
|
Runtime().api->ConnectionClose(conn);
|
|
}
|
|
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;
|
|
// 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);
|
|
}
|
|
|
|
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;
|
|
}
|