fix(quic): stop ~ClientQUIC aborting the process on teardown #8
5 changed files with 286 additions and 16 deletions
|
|
@ -275,6 +275,7 @@ The library includes tests covering:
|
||||||
- HTTP/3 external interop (`ShouldSend`) — live fetch from `cloudflare-quic.com:443`, exercises real TLS chain validation, mandatory control stream, peer-initiated unidi streams, and QPACK Huffman decoding
|
- HTTP/3 external interop (`ShouldSend`) — live fetch from `cloudflare-quic.com:443`, exercises real TLS chain validation, mandatory control stream, peer-initiated unidi streams, and QPACK Huffman decoding
|
||||||
- QUIC reliable streams (`ShouldSendRecieveQUICStream`)
|
- QUIC reliable streams (`ShouldSendRecieveQUICStream`)
|
||||||
- QUIC unreliable datagrams (`ShouldSendRecieveQUICDatagram`)
|
- QUIC unreliable datagrams (`ShouldSendRecieveQUICDatagram`)
|
||||||
|
- QUIC connection teardown (`ShouldSurviveConnectionChurn`) — 25 build-up/tear-down cycles with reader threads parked inside `RecieveSync`, the shape a reconnect loop has; the rest of the QUIC tests end in `std::_Exit(0)` and so never run a `ClientQUIC` destructor
|
||||||
- WebTransport echo (`ShouldEchoWebTransport`) — extended-CONNECT acceptance, draft-02 SETTINGS, bidi data stream framing (`WT_STREAM 0x41` + session-id varint), and byte-for-byte echo
|
- WebTransport echo (`ShouldEchoWebTransport`) — extended-CONNECT acceptance, draft-02 SETTINGS, bidi data stream framing (`WT_STREAM 0x41` + session-id varint), and byte-for-byte echo
|
||||||
- HTTP/1.1 wire format (`ShouldParseHTTP1`) — serialisation, incremental parsing fed one byte at a time, chunked bodies with trailers, pipelining, interim 1xx, HTTP/1.0 and HEAD framing, and the malformed inputs the parser is required to reject
|
- HTTP/1.1 wire format (`ShouldParseHTTP1`) — serialisation, incremental parsing fed one byte at a time, chunked bodies with trailers, pipelining, interim 1xx, HTTP/1.0 and HEAD framing, and the malformed inputs the parser is required to reject
|
||||||
- HTTP/1.1 round-trip (`ShouldSendRecieveHTTP1`) — routing, query strings, HEAD, 404 and a throwing handler
|
- HTTP/1.1 round-trip (`ShouldSendRecieveHTTP1`) — routing, query strings, HEAD, 404 and a throwing handler
|
||||||
|
|
|
||||||
|
|
@ -50,6 +50,47 @@ namespace {
|
||||||
return r;
|
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 FinalizeClose has run StreamClose. ~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.
|
||||||
|
constexpr std::chrono::milliseconds streamDrainTimeout{ 5000 };
|
||||||
|
|
||||||
// Encode an ALPN string into the wire format msquic expects: a length
|
// Encode an ALPN string into the wire format msquic expects: a length
|
||||||
// byte followed by the ASCII characters. Lifetime of the returned buffer
|
// byte followed by the ASCII characters. Lifetime of the returned buffer
|
||||||
// matches the caller's storage in `out`.
|
// matches the caller's storage in `out`.
|
||||||
|
|
@ -85,15 +126,29 @@ struct QUICStream::Impl {
|
||||||
// ref (wrapper / in-flight callback copy) drops.
|
// ref (wrapper / in-flight callback copy) drops.
|
||||||
std::shared_ptr<Impl>* selfRef = nullptr;
|
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 by FinalizeClose.
|
||||||
|
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);
|
||||||
|
|
||||||
// Detach the msquic handler, close the stream, and release the callback's
|
// Detach the msquic handler, close the stream, and release the callback's
|
||||||
// strong ref. Called once, from SHUTDOWN_COMPLETE (normal path) or the
|
// strong ref. Called once, from SHUTDOWN_COMPLETE (normal path) or the
|
||||||
// OpenStream error path — never racing another close of the same stream.
|
// OpenStream error path — never racing another close of the same stream.
|
||||||
static void FinalizeClose(Impl* self, HQUIC stream) {
|
static void FinalizeClose(Impl* self, HQUIC stream) {
|
||||||
Runtime().api->SetCallbackHandler(stream, nullptr, nullptr);
|
Runtime().api->SetCallbackHandler(stream, nullptr, nullptr);
|
||||||
Runtime().api->StreamClose(stream);
|
Runtime().api->StreamClose(stream);
|
||||||
|
std::shared_ptr<StreamRegistry> registry = std::move(self->registry);
|
||||||
std::shared_ptr<Impl>* ref = self->selfRef;
|
std::shared_ptr<Impl>* ref = self->selfRef;
|
||||||
self->selfRef = nullptr;
|
self->selfRef = nullptr;
|
||||||
delete ref;
|
delete ref;
|
||||||
|
// Last: a destructor blocked in WaitForDrain may run ConnectionClose
|
||||||
|
// the moment this hits zero, and msquic wants StreamClose first.
|
||||||
|
if (registry) registry->Remove();
|
||||||
}
|
}
|
||||||
|
|
||||||
static QUIC_STATUS QUIC_API Callback(HQUIC stream, void* ctx, QUIC_STREAM_EVENT* ev) {
|
static QUIC_STATUS QUIC_API Callback(HQUIC stream, void* ctx, QUIC_STREAM_EVENT* ev) {
|
||||||
|
|
@ -163,6 +218,8 @@ QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection)
|
||||||
{
|
{
|
||||||
impl->handle = handle;
|
impl->handle = handle;
|
||||||
impl->connection = connection;
|
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
|
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);
|
Runtime().api->SetCallbackHandler(handle, reinterpret_cast<void*>(&Impl::Callback), impl->selfRef);
|
||||||
}
|
}
|
||||||
|
|
@ -355,8 +412,38 @@ struct ClientQUIC::Impl {
|
||||||
// H3_MISSING_SETTINGS on the peer side.
|
// H3_MISSING_SETTINGS on the peer side.
|
||||||
std::deque<QUICStream> pendingStreams;
|
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;
|
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
|
||||||
|
// FinalizeClose, 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) {
|
static QUIC_STATUS QUIC_API Callback(HQUIC conn, void* ctx, QUIC_CONNECTION_EVENT* ev) {
|
||||||
auto* self = static_cast<Impl*>(ctx);
|
auto* self = static_cast<Impl*>(ctx);
|
||||||
switch (ev->Type) {
|
switch (ev->Type) {
|
||||||
|
|
@ -386,9 +473,16 @@ struct ClientQUIC::Impl {
|
||||||
self->closed = true;
|
self->closed = true;
|
||||||
}
|
}
|
||||||
self->cv.notify_all();
|
self->cv.notify_all();
|
||||||
if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0) {
|
// 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 FinalizeClose)
|
||||||
|
// by the time msquic reports the connection complete.
|
||||||
|
if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0
|
||||||
|
&& self->ClaimConnection() != nullptr) {
|
||||||
Runtime().api->ConnectionClose(conn);
|
Runtime().api->ConnectionClose(conn);
|
||||||
self->connection = nullptr;
|
|
||||||
}
|
}
|
||||||
return QUIC_STATUS_SUCCESS;
|
return QUIC_STATUS_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
@ -461,6 +555,12 @@ static HQUIC OpenClientConfiguration(const std::string& alpn, const QUICClientCr
|
||||||
QUIC_SETTINGS settings{};
|
QUIC_SETTINGS settings{};
|
||||||
settings.IsSet.IdleTimeoutMs = 1;
|
settings.IsSet.IdleTimeoutMs = 1;
|
||||||
settings.IdleTimeoutMs = 120'000;
|
settings.IdleTimeoutMs = 120'000;
|
||||||
|
// Bound the handshake separately from the (much longer) idle timeout.
|
||||||
|
// The constructor blocks until msquic reports the connection either up or
|
||||||
|
// shut down; without this, dialling a host that never answers leaves it
|
||||||
|
// waiting for the idle timeout — minutes rather than seconds.
|
||||||
|
settings.IsSet.HandshakeIdleTimeoutMs = 1;
|
||||||
|
settings.HandshakeIdleTimeoutMs = 10'000;
|
||||||
// Keep the connection alive across long idle gaps. msquic sends PING frames
|
// 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
|
// 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
|
// connection survives even while the app is blocked for a long time between
|
||||||
|
|
@ -521,11 +621,23 @@ ClientQUIC::ClientQUIC(const char* host, std::uint16_t port, std::string alpnIn,
|
||||||
throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast<unsigned>(s)));
|
throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast<unsigned>(s)));
|
||||||
}
|
}
|
||||||
|
|
||||||
std::unique_lock lk(impl->mtx);
|
QUIC_STATUS handshakeStatus = QUIC_STATUS_SUCCESS;
|
||||||
impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; });
|
{
|
||||||
if (!impl->connected) {
|
std::unique_lock lk(impl->mtx);
|
||||||
throw QUICException(std::format("QUIC handshake failed: 0x{:x}", static_cast<unsigned>(impl->shutdownStatus)));
|
// 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::ClientQUIC(std::string host, std::uint16_t port, std::string alpnIn, QUICClientCredentials creds)
|
||||||
|
|
@ -552,11 +664,7 @@ ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept
|
||||||
|
|
||||||
ClientQUIC::~ClientQUIC() {
|
ClientQUIC::~ClientQUIC() {
|
||||||
if (!impl) return;
|
if (!impl) return;
|
||||||
if (impl->connection) {
|
impl->CloseConnection();
|
||||||
Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
|
||||||
Runtime().api->ConnectionClose(impl->connection);
|
|
||||||
impl->connection = nullptr;
|
|
||||||
}
|
|
||||||
if (impl->configuration && impl->ownsConfiguration) {
|
if (impl->configuration && impl->ownsConfiguration) {
|
||||||
Runtime().api->ConfigurationClose(impl->configuration);
|
Runtime().api->ConfigurationClose(impl->configuration);
|
||||||
impl->configuration = nullptr;
|
impl->configuration = nullptr;
|
||||||
|
|
@ -564,8 +672,15 @@ ClientQUIC::~ClientQUIC() {
|
||||||
}
|
}
|
||||||
|
|
||||||
void ClientQUIC::Stop() {
|
void ClientQUIC::Stop() {
|
||||||
if (!impl || !impl->connection) return;
|
if (!impl) return;
|
||||||
Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0);
|
// 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) {
|
QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
||||||
|
|
@ -573,6 +688,11 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
||||||
QUICStream stream;
|
QUICStream stream;
|
||||||
stream.impl = std::make_shared<QUICStream::Impl>();
|
stream.impl = std::make_shared<QUICStream::Impl>();
|
||||||
stream.impl->connection = this;
|
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);
|
stream.impl->selfRef = new std::shared_ptr<QUICStream::Impl>(stream.impl);
|
||||||
QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional
|
QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional
|
||||||
? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL
|
? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL
|
||||||
|
|
@ -583,6 +703,9 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) {
|
||||||
if (QUIC_FAILED(s)) {
|
if (QUIC_FAILED(s)) {
|
||||||
delete stream.impl->selfRef;
|
delete stream.impl->selfRef;
|
||||||
stream.impl->selfRef = nullptr;
|
stream.impl->selfRef = nullptr;
|
||||||
|
// No msquic stream exists, so FinalizeClose is not the right undo.
|
||||||
|
std::shared_ptr<StreamRegistry> registry = std::move(stream.impl->registry);
|
||||||
|
registry->Remove();
|
||||||
throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
|
throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
|
||||||
}
|
}
|
||||||
stream.handle = streamHandle;
|
stream.handle = streamHandle;
|
||||||
|
|
@ -658,3 +781,8 @@ std::vector<char> ClientQUIC::RecieveDatagramSync() {
|
||||||
}
|
}
|
||||||
|
|
||||||
HQUIC ClientQUIC::GetHandle() const { return impl ? impl->connection : nullptr; }
|
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;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -149,9 +149,11 @@ namespace Crafter {
|
||||||
// via OnStream()).
|
// via OnStream()).
|
||||||
// - Unreliable, unordered datagrams (SendDatagram() / OnDatagram()).
|
// - Unreliable, unordered datagrams (SendDatagram() / OnDatagram()).
|
||||||
//
|
//
|
||||||
// Lifetime: ~ClientQUIC closes the connection. Streams obtained from
|
// Lifetime: ~ClientQUIC shuts the connection down and closes it, waiting
|
||||||
// OpenStream() are scoped to the connection and must be destroyed (or
|
// (bounded — 5s) for every stream on it to finish closing first, since
|
||||||
// moved out) before the ClientQUIC.
|
// QUICStream::Stop only initiates a shutdown that completes asynchronously.
|
||||||
|
// Streams outliving the connection are therefore safe to hold, though they
|
||||||
|
// are dead once it goes; prefer destroying or moving them out first.
|
||||||
//
|
//
|
||||||
// Browser build: the only QUIC-shaped API the browser exposes is
|
// Browser build: the only QUIC-shaped API the browser exposes is
|
||||||
// WebTransport, which is HTTP/3-based and reached at a fixed URL. Here:
|
// WebTransport, which is HTTP/3-based and reached at a fixed URL. Here:
|
||||||
|
|
@ -178,6 +180,8 @@ namespace Crafter {
|
||||||
|
|
||||||
// Client constructor: connects to host:port using QUIC. ALPN must
|
// Client constructor: connects to host:port using QUIC. ALPN must
|
||||||
// match the listener. Throws QUICException on connect failure.
|
// match the listener. Throws QUICException on connect failure.
|
||||||
|
// Blocks for at most the handshake idle timeout (10s), so a host that
|
||||||
|
// never answers fails in seconds rather than at the idle timeout.
|
||||||
ClientQUIC(const char* host, std::uint16_t port, std::string alpn,
|
ClientQUIC(const char* host, std::uint16_t port, std::string alpn,
|
||||||
QUICClientCredentials creds = {});
|
QUICClientCredentials creds = {});
|
||||||
ClientQUIC(std::string host, std::uint16_t port, std::string alpn,
|
ClientQUIC(std::string host, std::uint16_t port, std::string alpn,
|
||||||
|
|
|
||||||
|
|
@ -149,6 +149,7 @@ extern "C" Configuration CrafterBuildProject(std::span<const std::string_view> a
|
||||||
cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg });
|
cfg.AddTest("ShouldSendRecieveQUICDatagram").Dependencies({ &cfg });
|
||||||
cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg });
|
cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg });
|
||||||
cfg.AddTest("ShouldSurviveAbuseHTTP1").Dependencies({ &cfg });
|
cfg.AddTest("ShouldSurviveAbuseHTTP1").Dependencies({ &cfg });
|
||||||
|
cfg.AddTest("ShouldSurviveConnectionChurn").Dependencies({ &cfg });
|
||||||
}
|
}
|
||||||
|
|
||||||
return cfg;
|
return cfg;
|
||||||
|
|
|
||||||
136
tests/ShouldSurviveConnectionChurn/main.cpp
Normal file
136
tests/ShouldSurviveConnectionChurn/main.cpp
Normal file
|
|
@ -0,0 +1,136 @@
|
||||||
|
//SPDX-License-Identifier: LGPL-3.0-only
|
||||||
|
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
||||||
|
|
||||||
|
// Regression test for the connection-teardown race in ~ClientQUIC.
|
||||||
|
//
|
||||||
|
// QUICStream::Stop only *initiates* a graceful shutdown; the actual
|
||||||
|
// MsQuicStreamClose happens later, from the terminal SHUTDOWN_COMPLETE
|
||||||
|
// callback on an msquic worker thread. ~ClientQUIC used to call
|
||||||
|
// MsQuicConnectionClose straight away, so a connection could be closed while
|
||||||
|
// streams belonging to it were still open on the msquic side — which trips an
|
||||||
|
// msquic bugcheck and aborts the process.
|
||||||
|
//
|
||||||
|
// Nothing else in the suite exercises this: every other QUIC test ends in
|
||||||
|
// std::_Exit(0) and so never runs a ClientQUIC destructor. The shape that
|
||||||
|
// breaks is a reconnect loop — build a connection, tear it down, build
|
||||||
|
// another — which is exactly what this test does, `cycles` times.
|
||||||
|
//
|
||||||
|
// The blocked readers matter. Each inbound stream gets a thread parked inside
|
||||||
|
// RecieveSync for the life of the connection (the shape a long-lived
|
||||||
|
// streaming client has), so teardown happens with readers still inside the
|
||||||
|
// receive path. An otherwise identical loop whose handler reads once and
|
||||||
|
// returns does not reproduce.
|
||||||
|
|
||||||
|
import Crafter.Network;
|
||||||
|
import Crafter.Thread;
|
||||||
|
import std;
|
||||||
|
using namespace Crafter;
|
||||||
|
|
||||||
|
namespace {
|
||||||
|
constexpr int cycles = 25;
|
||||||
|
constexpr int streamsPerConnection = 3;
|
||||||
|
}
|
||||||
|
|
||||||
|
int main() {
|
||||||
|
ThreadPool::Start();
|
||||||
|
|
||||||
|
// On a build where a teardown deadlocks rather than aborts, fail the test
|
||||||
|
// instead of hanging the whole suite.
|
||||||
|
std::thread watchdog([] {
|
||||||
|
std::this_thread::sleep_for(std::chrono::seconds(90));
|
||||||
|
std::println("timed out — a connection teardown never completed");
|
||||||
|
std::cout.flush();
|
||||||
|
std::_Exit(1);
|
||||||
|
});
|
||||||
|
watchdog.detach();
|
||||||
|
|
||||||
|
for (int i = 0; i < cycles; ++i) {
|
||||||
|
const std::uint16_t port = static_cast<std::uint16_t>(19200 + i);
|
||||||
|
const std::string alpn = "churn/1";
|
||||||
|
|
||||||
|
QUICServerCredentials serverCreds;
|
||||||
|
serverCreds.selfSigned = true;
|
||||||
|
|
||||||
|
std::mutex serverMutex;
|
||||||
|
std::vector<QUICStream> serverStreams;
|
||||||
|
ClientQUIC* accepted = nullptr;
|
||||||
|
|
||||||
|
auto listener = std::make_unique<ListenerQUIC>(
|
||||||
|
port, alpn, serverCreds, [&](ClientQUIC* peer) {
|
||||||
|
std::lock_guard lock(serverMutex);
|
||||||
|
accepted = peer;
|
||||||
|
for (int s = 0; s < streamsPerConnection; ++s) {
|
||||||
|
QUICStream stream = peer->OpenStream(/*unidirectional=*/true);
|
||||||
|
const char payload[] = "hello";
|
||||||
|
stream.SendSync(payload, sizeof(payload) - 1, /*finish=*/false);
|
||||||
|
serverStreams.push_back(std::move(stream));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
listener->ListenAsyncAsync();
|
||||||
|
|
||||||
|
QUICClientCredentials clientCreds;
|
||||||
|
clientCreds.insecureNoServerValidation = true;
|
||||||
|
|
||||||
|
{
|
||||||
|
ClientQUIC client(std::string("localhost"), port, alpn, clientCreds);
|
||||||
|
|
||||||
|
std::mutex clientMutex;
|
||||||
|
std::condition_variable clientCv;
|
||||||
|
std::vector<std::thread> readers;
|
||||||
|
int started = 0;
|
||||||
|
|
||||||
|
client.OnStream([&](QUICStream stream) {
|
||||||
|
auto shared = std::make_shared<QUICStream>(std::move(stream));
|
||||||
|
std::thread reader([shared] {
|
||||||
|
try {
|
||||||
|
while (true) (void)shared->RecieveSync();
|
||||||
|
} catch (...) {
|
||||||
|
// Connection closed.
|
||||||
|
}
|
||||||
|
});
|
||||||
|
std::lock_guard lock(clientMutex);
|
||||||
|
readers.push_back(std::move(reader));
|
||||||
|
++started;
|
||||||
|
clientCv.notify_all();
|
||||||
|
});
|
||||||
|
|
||||||
|
{
|
||||||
|
std::unique_lock lock(clientMutex);
|
||||||
|
clientCv.wait_for(lock, std::chrono::seconds(5), [&] {
|
||||||
|
return started == streamsPerConnection;
|
||||||
|
});
|
||||||
|
if (started != streamsPerConnection) {
|
||||||
|
std::println("cycle {}: only {} of {} streams arrived",
|
||||||
|
i, started, streamsPerConnection);
|
||||||
|
return 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
client.Stop();
|
||||||
|
for (std::thread& reader : readers) {
|
||||||
|
if (reader.joinable()) reader.join();
|
||||||
|
}
|
||||||
|
// `client` is destroyed here, once the readers have unwound.
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
std::lock_guard lock(serverMutex);
|
||||||
|
// Deliberately the other order from the client side: destroy the
|
||||||
|
// connection while the app still holds its stream wrappers, so
|
||||||
|
// ~ClientQUIC has to cope with streams that are open as far as
|
||||||
|
// msquic is concerned.
|
||||||
|
delete accepted;
|
||||||
|
accepted = nullptr;
|
||||||
|
serverStreams.clear();
|
||||||
|
}
|
||||||
|
listener->Stop();
|
||||||
|
listener.reset();
|
||||||
|
}
|
||||||
|
|
||||||
|
std::println("survived {} connection teardowns", cycles);
|
||||||
|
std::cout.flush();
|
||||||
|
// Skip the static-dtor cleanup: msquic's RegistrationClose blocks until
|
||||||
|
// every connection it ever opened is fully drained, which the suite does
|
||||||
|
// not need to wait on. Everything this test asserts has already happened.
|
||||||
|
std::_Exit(0);
|
||||||
|
}
|
||||||
Loading…
Reference in a new issue