Merge pull request 'fix(quic): close stream handles whose async StreamStart lost the teardown race' (#10) from claude/issue-9 into master

implementations/Crafter.Network-ClientQUIC.cpp

@@ -52,7 +52,7 @@ namespace {
 
     // 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
+    // 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
@@ -89,6 +89,14 @@ namespace {
     // 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
@@ -106,7 +114,6 @@ namespace {
 
 // ---------------- QUICStream::Impl ----------------
 struct QUICStream::Impl {
-    HQUIC handle = nullptr;
     ClientQUIC* connection = nullptr;
 
     std::mutex mtx;
@@ -128,7 +135,7 @@ struct QUICStream::Impl {
 
     // The owning connection's open-stream count. Incremented where the msquic
     // handler is installed (QUICStream's handle constructor / OpenStream),
-    // decremented by FinalizeClose.
+    // decremented once the handle has been closed.
     std::shared_ptr<StreamRegistry> registry;
 
     // The connection's registry. Defined below ClientQUIC::Impl (which owns
@@ -136,19 +143,73 @@ struct QUICStream::Impl {
     // 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
+    // ---- msquic handle ownership ----------------------------------------
-    // strong ref. Called once, from SHUTDOWN_COMPLETE (normal path) or the
+    // StreamClose is "equivalent to free": any other msquic call on the handle
-    // OpenStream error path — never racing another close of the same stream.
+    // that overlaps it is a use-after-free, which msquic reports as a bugcheck
-    static void FinalizeClose(Impl* self, HQUIC stream) {
+    // on whichever thread trips over the wreckage. The handle therefore has a
-        Runtime().api->SetCallbackHandler(stream, nullptr, nullptr);
+    // refcount, and whoever drops the last ref closes it. There are two kinds
-        Runtime().api->StreamClose(stream);
+    // of holder:
-        std::shared_ptr<StreamRegistry> registry = std::move(self->registry);
+    //
-        std::shared_ptr<Impl>* ref = self->selfRef;
+    //   - The msquic callback holds exactly one ref for the stream's lifetime,
-        self->selfRef = nullptr;
+    //     released by Retire() once SHUTDOWN_COMPLETE has been delivered (or by
-        delete ref;
+    //     the OpenStream error paths, where no callback will ever run).
-        // Last: a destructor blocked in WaitForDrain may run ConnectionClose
+    //   - Every app thread inside an msquic call on the handle holds one for
-        // the moment this hits zero, and msquic wants StreamClose first.
+    //     the duration of that call (AcquireHandle / ReleaseHandle).
-        if (registry) registry->Remove();
+    //
+    // 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) {
@@ -200,9 +261,12 @@ struct QUICStream::Impl {
                     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();
-                FinalizeClose(self, stream);
+                self->Retire();
                 return QUIC_STATUS_SUCCESS;
             }
             default:
@@ -216,7 +280,7 @@ QUICStream::QUICStream() = default;
 QUICStream::QUICStream(HQUIC handle, ClientQUIC* connection)
     : handle(handle), connection(connection), impl(std::make_shared<Impl>())
 {
-    impl->handle = handle;
+    impl->SetHandle(handle);   // handleRefs already accounts for the callback
     impl->connection = connection;
     impl->registry = Impl::RegistryOf(connection);
     if (impl->registry) impl->registry->Add();
@@ -253,29 +317,34 @@ QUICStream::~QUICStream() {
 
 void QUICStream::Stop() {
     if (!handle) return;
-    // Only INITIATE a graceful shutdown here; the async SHUTDOWN_COMPLETE
+    handle = nullptr;   // the wrapper is done with the stream either way
-    // callback performs the actual StreamClose + ref release. The Impl is a
+    if (!impl) return;
-    // shared_ptr held by the callback's selfRef, so it safely outlives this
+    // Only INITIATE the shutdown here; the async SHUTDOWN_COMPLETE callback
-    // wrapper until that callback runs. If SHUTDOWN_COMPLETE already fired,
+    // releases the handle, and the Impl is a shared_ptr held by the callback's
-    // msquic has closed the stream and calling StreamShutdown again would trip
+    // selfRef, so it safely outlives this wrapper until that callback runs.
-    // a bugcheck — skip it.
+    // A null handle means SHUTDOWN_COMPLETE already fired and msquic has closed
-    bool alreadyClosed = false;
+    // the stream — shutting it down again would be a use-after-free.
-    if (impl) {
+    HQUIC live = impl->AcquireHandle();
-        std::lock_guard lk(impl->mtx);
+    if (!live) return;
-        alreadyClosed = impl->shutdownComplete;
+    // 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.
-    if (!alreadyClosed) {
+    // Without it a bidirectional stream whose peer keeps its send side open
-        Runtime().api->StreamShutdown(handle, QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0);
+    // never reaches SHUTDOWN_COMPLETE, so its handle is never closed and the
-    }
+    // connection can never finish tearing down.
-    handle = nullptr;
+    Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0);
-    if (impl) {
+    if (canReceive) {
-        std::lock_guard lk(impl->mtx);
+        Runtime().api->StreamShutdown(live, QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, 0);
-        impl->handle = nullptr;
     }
+    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{};
@@ -286,8 +355,15 @@ void QUICStream::SendSync(const void* buffer, std::uint32_t size, bool finish) {
         impl->sendInFlight = true;
     }
     QUIC_SEND_FLAGS flags = finish ? QUIC_SEND_FLAG_FIN : QUIC_SEND_FLAG_NONE;
-    QUIC_STATUS s = Runtime().api->StreamSend(handle, &quicBuf, 1, flags, copy);
+    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)));
     }
@@ -380,9 +456,15 @@ void QUICStream::PrependReceived(std::vector<char> bytes) {
 
 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(handle, QUIC_PARAM_STREAM_ID, &size, &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)));
@@ -438,7 +520,7 @@ struct ClientQUIC::Impl {
         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.
+        // 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);
@@ -478,8 +560,8 @@ struct ClientQUIC::Impl {
                 // 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)
+                // has already been shut down (and so closed) by the time
-                // by the time msquic reports the connection complete.
+                // msquic reports the connection complete.
                 if (ev->SHUTDOWN_COMPLETE.AppCloseInProgress == 0
                     && self->ClaimConnection() != nullptr) {
                     Runtime().api->ConnectionClose(conn);
@@ -660,6 +742,17 @@ ClientQUIC::ClientQUIC(ClientQUIC&& other) noexcept
 
 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);
@@ -699,26 +792,36 @@ 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.
+        // 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->handle = streamHandle;
+    stream.impl->SetHandle(streamHandle);
     if (unidirectional) {
         // We initiated the unidi stream: we send, peer reads.
         stream.canSend = true;
         stream.canReceive = false;
     }
-    s = Runtime().api->StreamStart(streamHandle, QUIC_STREAM_START_FLAG_NONE);
+    // 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)) {
-        // Handler is registered; detach + close so no callback fires on the
+        // Synchronous failure: the stream never started, so no callback will
-        // failed stream, and clear the wrapper handle so ~QUICStream is a no-op.
+        // ever run. Release the callback's ref by hand, which closes the
-        QUICStream::Impl::FinalizeClose(stream.impl.get(), streamHandle);
+        // handle, and clear the wrapper's so ~QUICStream is a no-op.
         stream.handle = nullptr;
-        stream.impl->handle = nullptr;
+        stream.impl->Retire();
         throw QUICException(std::format("StreamStart failed: 0x{:x}", static_cast<unsigned>(s)));
     }
     return stream;

interfaces/Crafter.Network-ClientQUIC.cppm

@@ -158,7 +158,13 @@ namespace Crafter {
         std::uint64_t GetStreamId() const;
 #endif
 
-        // Cleanly shut down the stream (both directions).
+        // Shut the stream down in both directions: our send side gracefully
+        // (the peer sees a FIN), the receive side abortively. Aborting the
+        // receive side matters — a bidirectional stream whose peer keeps its
+        // send side open otherwise never completes its shutdown, and so never
+        // releases its transport handle. Returns immediately; completion is
+        // asynchronous. Called by the destructor, so an explicit call is only
+        // needed to shut a stream down ahead of its wrapper going away.
         void Stop();
 
     private:

tests/ShouldSurviveConnectionChurn/main.cpp

@@ -1,25 +1,38 @@
 //SPDX-License-Identifier: LGPL-3.0-only
 //SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
 
-// Regression test for the connection-teardown race in ~ClientQUIC.
+// Regression test for connection teardown on the client side.
 //
-// QUICStream::Stop only *initiates* a graceful shutdown; the actual
+// QUICStream::Stop only *initiates* a shutdown; the msquic handle is closed
-// MsQuicStreamClose happens later, from the terminal SHUTDOWN_COMPLETE
+// later, off the terminal SHUTDOWN_COMPLETE callback on a worker thread. Two
-// callback on an msquic worker thread. ~ClientQUIC used to call
+// things used to go wrong with that, and both need a reconnect loop to show
-// MsQuicConnectionClose straight away, so a connection could be closed while
+// up, which is what this test is:
-// 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
+//  1. ~ClientQUIC closed the connection straight away, so a connection could
-// std::_Exit(0) and so never runs a ClientQUIC destructor. The shape that
+//     be closed while streams belonging to it were still open on the msquic
-// breaks is a reconnect loop — build a connection, tear it down, build
+//     side — an msquic bugcheck, which aborts the process.
-// another — which is exactly what this test does, `cycles` times.
 //
-// The blocked readers matter. Each inbound stream gets a thread parked inside
+//  2. StreamStart is asynchronous. If the connection was shut down before the
+//     queued start ran, msquic failed the start and left the stream neither
+//     started nor shut down, so SHUTDOWN_COMPLETE never arrived and the handle
+//     was never closed. The registration's rundown at exit() then blocked
+//     forever waiting for it.
+//
+// The test therefore ends by falling off the end of main() rather than with
+// std::_Exit — the static MsQuicRuntime destructor's RegistrationClose is
+// where a leaked handle shows up, so skipping it would skip half the point.
+//
+// Two things about the shape matter:
+//
+//  - Blocked readers. 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
+//    receive path.
-// returns does not reproduce.
+//
+//  - A concurrent sender. A client-opened bidirectional control stream with a
+//    thread sending on it across the teardown, which is the shape a client
+//    with periodic acks or keepalives has. It is the send racing the shutdown
+//    that reaches (2); a loop without it passes on the broken build.
 
 import Crafter.Network;
 import Crafter.Thread;
@@ -57,6 +70,20 @@ int main() {
 
         auto listener = std::make_unique<ListenerQUIC>(
             port, alpn, serverCreds, [&](ClientQUIC* peer) {
+                // Drain whatever the client opens toward us. Without this the
+                // client's control-stream sends would block on flow control
+                // and the test would wedge on its own account rather than on
+                // anything the library did.
+                peer->OnStream([](QUICStream inbound) {
+                    auto shared = std::make_shared<QUICStream>(std::move(inbound));
+                    std::thread([shared] {
+                        try {
+                            while (true) (void)shared->RecieveSync();
+                        } catch (...) {
+                            // Connection closed.
+                        }
+                    }).detach();
+                });
                 std::lock_guard lock(serverMutex);
                 accepted = peer;
                 for (int s = 0; s < streamsPerConnection; ++s) {
@@ -106,22 +133,40 @@ int main() {
                 }
             }
 
+            // The sends keep going while the connection is torn down
+            // underneath them.
+            QUICStream control = client.OpenStream(/*unidirectional=*/false);
+            std::atomic<bool> sending{ true };
+            std::thread sender([&] {
+                while (sending) {
+                    try {
+                        const char tick[] = "ack";
+                        control.SendSync(tick, sizeof(tick) - 1, /*finish=*/false);
+                    } catch (...) {
+                        return;   // connection went away mid-send
+                    }
+                    std::this_thread::sleep_for(std::chrono::milliseconds(2));
+                }
+            });
+
             client.Stop();
+            sending = false;
+            if (sender.joinable()) sender.join();
             for (std::thread& reader : readers) {
                 if (reader.joinable()) reader.join();
             }
-            // `client` is destroyed here, once the readers have unwound.
+            // `control` and then `client` are destroyed here, once the sender
+            // and readers have unwound.
         }
 
         {
             std::lock_guard lock(serverMutex);
-            // Deliberately the other order from the client side: destroy the
+            // Deliberately the other order from the client side: shut the
-            // connection while the app still holds its stream wrappers, so
+            // connection down while the app still holds its stream wrappers,
-            // ~ClientQUIC has to cope with streams that are open as far as
+            // so teardown has to cope with streams that are open as far as
             // msquic is concerned.
-            delete accepted;
-            accepted = nullptr;
             serverStreams.clear();
+            if (accepted) accepted->Stop();
         }
         listener->Stop();
         listener.reset();
@@ -129,8 +174,9 @@ int main() {
 
     std::println("survived {} connection teardowns", cycles);
     std::cout.flush();
-    // Skip the static-dtor cleanup: msquic's RegistrationClose blocks until
+    // Deliberately no std::_Exit: returning from main runs the static
-    // every connection it ever opened is fully drained, which the suite does
+    // MsQuicRuntime destructor, whose RegistrationClose blocks until every
-    // not need to wait on. Everything this test asserts has already happened.
+    // handle opened under the registration has been closed. That wait is the
-    std::_Exit(0);
+    // assertion for (2) above.
+    return 0;
 }