diff --git a/implementations/Crafter.Network-ClientQUIC.cpp b/implementations/Crafter.Network-ClientQUIC.cpp index 323a45b..dbe16df 100644 --- a/implementations/Crafter.Network-ClientQUIC.cpp +++ b/implementations/Crafter.Network-ClientQUIC.cpp @@ -50,6 +50,47 @@ namespace { 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 // byte followed by the ASCII characters. Lifetime of the returned buffer // matches the caller's storage in `out`. @@ -85,15 +126,29 @@ struct QUICStream::Impl { // ref (wrapper / in-flight callback copy) drops. std::shared_ptr* 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 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 RegistryOf(ClientQUIC* connection); + // Detach the msquic handler, close the stream, and release the callback's // strong ref. Called once, from SHUTDOWN_COMPLETE (normal path) or the // OpenStream error path — never racing another close of the same stream. static void FinalizeClose(Impl* self, HQUIC stream) { Runtime().api->SetCallbackHandler(stream, nullptr, nullptr); Runtime().api->StreamClose(stream); + std::shared_ptr registry = std::move(self->registry); std::shared_ptr* ref = self->selfRef; self->selfRef = nullptr; 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) { @@ -163,6 +218,8 @@ QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection) { impl->handle = handle; impl->connection = connection; + impl->registry = Impl::RegistryOf(connection); + if (impl->registry) impl->registry->Add(); impl->selfRef = new std::shared_ptr(impl); // strong ref owned by the callback Runtime().api->SetCallbackHandler(handle, reinterpret_cast(&Impl::Callback), impl->selfRef); } @@ -355,8 +412,38 @@ struct ClientQUIC::Impl { // H3_MISSING_SETTINGS on the peer side. std::deque pendingStreams; + // Streams currently open on this connection. Shared with each QUICStream + // so the count survives a stream that outlives us. + std::shared_ptr streams = std::make_shared(); + 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) { auto* self = static_cast(ctx); switch (ev->Type) { @@ -386,9 +473,16 @@ struct ClientQUIC::Impl { self->closed = true; } 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); - self->connection = nullptr; } return QUIC_STATUS_SUCCESS; } @@ -521,11 +615,21 @@ ClientQUIC::ClientQUIC(const char* host, std::uint16_t port, std::string alpnIn, throw QUICException(std::format("ConnectionStart failed: 0x{:x}", static_cast(s))); } - std::unique_lock lk(impl->mtx); - impl->cv.wait(lk, [&]{ return impl->connected || impl->closed; }); - if (!impl->connected) { - throw QUICException(std::format("QUIC handshake failed: 0x{:x}", static_cast(impl->shutdownStatus))); + QUIC_STATUS handshakeStatus = QUIC_STATUS_SUCCESS; + { + std::unique_lock lk(impl->mtx); + 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(handshakeStatus))); } ClientQUIC::ClientQUIC(std::string host, std::uint16_t port, std::string alpnIn, QUICClientCredentials creds) @@ -552,11 +656,7 @@ ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept ClientQUIC::~ClientQUIC() { if (!impl) return; - if (impl->connection) { - Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0); - Runtime().api->ConnectionClose(impl->connection); - impl->connection = nullptr; - } + impl->CloseConnection(); if (impl->configuration && impl->ownsConfiguration) { Runtime().api->ConfigurationClose(impl->configuration); impl->configuration = nullptr; @@ -564,8 +664,15 @@ ClientQUIC::~ClientQUIC() { } void ClientQUIC::Stop() { - if (!impl || !impl->connection) return; - Runtime().api->ConnectionShutdown(impl->connection, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0); + 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) { @@ -573,6 +680,11 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) { QUICStream stream; stream.impl = std::make_shared(); 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(stream.impl); QUIC_STREAM_OPEN_FLAGS openFlags = unidirectional ? QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL @@ -583,6 +695,9 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) { if (QUIC_FAILED(s)) { delete stream.impl->selfRef; stream.impl->selfRef = nullptr; + // No msquic stream exists, so FinalizeClose is not the right undo. + std::shared_ptr registry = std::move(stream.impl->registry); + registry->Remove(); throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast(s))); } stream.handle = streamHandle; @@ -658,3 +773,8 @@ std::vector ClientQUIC::RecieveDatagramSync() { } HQUIC ClientQUIC::GetHandle() const { return impl ? impl->connection : nullptr; } + +std::shared_ptr QUICStream::Impl::RegistryOf(ClientQUIC* connection) { + if (!connection || !connection->impl) return nullptr; + return connection->impl->streams; +} diff --git a/interfaces/Crafter.Network-ClientQUIC.cppm b/interfaces/Crafter.Network-ClientQUIC.cppm index bf35cbd..6068922 100644 --- a/interfaces/Crafter.Network-ClientQUIC.cppm +++ b/interfaces/Crafter.Network-ClientQUIC.cppm @@ -149,9 +149,11 @@ namespace Crafter { // via OnStream()). // - Unreliable, unordered datagrams (SendDatagram() / OnDatagram()). // - // Lifetime: ~ClientQUIC closes the connection. Streams obtained from - // OpenStream() are scoped to the connection and must be destroyed (or - // moved out) before the ClientQUIC. + // Lifetime: ~ClientQUIC shuts the connection down and closes it, waiting + // (bounded — 5s) for every stream on it to finish closing first, since + // 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 // WebTransport, which is HTTP/3-based and reached at a fixed URL. Here: