fix(quic): let the streams already stopped finish before ~ClientQUIC closes

~ClientQUIC shut the connection down first and drained its streams
after. msquic aborts every stream with the connection, and with them
whatever it had not sent yet, so a server that replied with
SendAsync(..., finish = true), let the stream go and deleted the
connection lost the reply whenever the shutdown, queued right behind
the send, reached the worker before the flush that would have sent it.
The interface promised the opposite: that the destructor waits for
every stream "to finish closing first".

Found with Crafter.Debug, whose session answers every request still
waiting when it ends and then deletes its connections: the follower
it told "shutting down" missed the reply in 4 runs of 30.

Streams now count as finishing from QUICStream::Stop to their
SHUTDOWN_COMPLETE, in the registry beside the open count, and the
destructor waits for that count to reach zero before the shutdown.
Only streams that can send are counted, since a stream that cannot has
nothing on its way. A stopped stream finishes once the peer has
acknowledged its FIN, about a round trip; the wait is bounded at 1 s
so a peer that stopped answering, or one that never reads, holds the
destructor no longer than that, and a peer closing meanwhile aborts the
streams, which ends the wait too. Streams their owner still holds are
not waited for: they are aborted as before.

Measured with ShouldFinishStoppedStreamsQUIC: before, 0 of 20 replies of
64 KiB arrived; after, 20 of 20, with the slowest delete at 1 ms.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

implementations/Crafter.Network-ClientQUIC.cpp

@@ -65,25 +65,41 @@ namespace {
         std::mutex mtx;
         std::condition_variable cv;
         int open = 0;
+        // Of those, the ones whose owner stopped them gracefully: what was
+        // sent on them, the FIN last, is still on its way to the peer.
+        int finishing = 0;
 
         void Add() {
             std::lock_guard lk(mtx);
             ++open;
         }
 
-        void Remove() {
+        // `wasFinishing`: the stream had been counted by Finishing().
+        void Remove(bool wasFinishing) {
             {
                 std::lock_guard lk(mtx);
                 --open;
+                if (wasFinishing) --finishing;
             }
             cv.notify_all();
         }
 
+        void Finishing() {
+            std::lock_guard lk(mtx);
+            ++finishing;
+        }
+
         // 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; });
         }
+
+        // False if the timeout expired with stopped streams still finishing.
+        bool WaitForFinishing(std::chrono::milliseconds timeout) {
+            std::unique_lock lk(mtx);
+            return cv.wait_for(lk, timeout, [&]{ return finishing == 0; });
+        }
     };
 
     // ClientQUIC::OnClosed's state, shared by the ClientQUIC and whoever
@@ -175,6 +191,18 @@ namespace {
     // stops a peer that is gone from holding the destructor that long.
     constexpr std::chrono::milliseconds closeNotifyTimeout{ 1000 };
 
+    // Bound on how long ~ClientQUIC first waits for the streams their owners
+    // stopped to finish. ConnectionShutdown aborts every stream, and with it
+    // whatever msquic still holds of theirs: a server that replied with
+    // SendAsync(..., finish = true), let the stream go and then deleted the
+    // connection lost the reply whenever msquic had not sent it yet, which
+    // the shutdown, queued behind the send, made likely (4 runs in 30 of a
+    // Crafter.Debug test lost the reply its session gave on shutdown). A
+    // stopped stream finishes once the peer has acknowledged its FIN, about
+    // a round trip; the bound only stops a peer that no longer answers, or
+    // one that never reads, from holding the destructor longer.
+    constexpr std::chrono::milliseconds finishTimeout{ 1000 };
+
     // Encode an ALPN string into the wire format msquic expects: a length
     // byte followed by the ASCII characters. Lifetime of the returned buffer
     // matches the caller's storage in `out`.
@@ -459,6 +487,9 @@ struct QUICStream::Impl {
     // handler is installed (QUICStream's handle constructor / OpenStream),
     // decremented once the handle has been closed.
     std::shared_ptr<StreamRegistry> registry;
+    // Counted in the registry's `finishing` since Stop() shut the send side
+    // down gracefully; under handleMtx.
+    bool finishing = false;
 
     // The connection's registry. Defined below ClientQUIC::Impl (which owns
     // it); a member of QUICStream::Impl rather than a free function so it
@@ -497,11 +528,13 @@ struct QUICStream::Impl {
 
     void ReleaseHandle() {
         HQUIC toClose = nullptr;
+        bool wasFinishing = false;
         {
             std::lock_guard lk(handleMtx);
             if (--handleRefs > 0) return;
             toClose = handle;
             handle = nullptr;
+            wasFinishing = finishing;
         }
         // Last ref: nobody else can be inside an msquic call on this handle,
         // and AcquireHandle will hand it out no more.
@@ -513,7 +546,21 @@ struct QUICStream::Impl {
         // 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();
+        if (reg) reg->Remove(wasFinishing);
+    }
+
+    // Stop()'s mark, with a handle ref held so the stream cannot be retired
+    // meanwhile: ~ClientQUIC lets the stream finish before it closes the
+    // connection. Once per stream.
+    void MarkFinishing() {
+        std::shared_ptr<StreamRegistry> reg;
+        {
+            std::lock_guard lk(handleMtx);
+            if (finishing || retired) return;
+            finishing = true;
+            reg = registry;
+        }
+        if (reg) reg->Finishing();
     }
 
     // Publish the msquic handle. Called once, while the stream is still local
@@ -709,6 +756,11 @@ void QUICStream::Stop() {
     // the stream — shutting it down again would be a use-after-free.
     HQUIC live = impl->AcquireHandle();
     if (!live) return;
+    // What was sent, and the FIN, still has to reach the peer: counted until
+    // SHUTDOWN_COMPLETE, so ~ClientQUIC waits for it before closing the
+    // connection, which would abort it. A stream that cannot send has
+    // nothing on its way.
+    if (canSend) impl->MarkFinishing();
     // 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
@@ -974,14 +1026,18 @@ struct ClientQUIC::Impl {
         return claimed;
     }
 
-    // Shut the connection down and close it, waiting (bounded) for every
+    // Shut the connection down and close it: first waiting (bounded) for
-    // stream on it to reach StreamClose first, and then for the shutdown to
+    // the streams already stopped to finish, so what was sent on them
-    // complete, so the peer is told. No-op if someone else already claimed
+    // arrives; then for every stream on it to reach StreamClose; then for
-    // the handle. Never holds `mtx` across an msquic call: the connection
+    // the shutdown to complete, so the peer is told. No-op if someone else
-    // callback takes it too.
+    // 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;
+        // See finishTimeout. A peer that closes meanwhile aborts them, which
+        // ends the wait as well.
+        streams->WaitForFinishing(finishTimeout);
         // 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);
@@ -1287,7 +1343,7 @@ QUICStream ClientQUIC::OpenStream(bool unidirectional) {
         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();
+        registry->Remove(false);
         throw QUICException(std::format("StreamOpen failed: 0x{:x}", static_cast<unsigned>(s)));
     }
     stream.handle = streamHandle;

interfaces/Crafter.Network-ClientQUIC.cppm

@@ -246,7 +246,8 @@ namespace Crafter {
         // asynchronous. Called by the destructor, so an explicit call is only
         // needed to shut a stream down ahead of its wrapper going away.
         // Natively it also unregisters OnRecieve's callbacks (see there);
-        // sends already made still go out, ahead of the FIN.
+        // sends already made still go out, ahead of the FIN, and
+        // ~ClientQUIC waits (bounded) for them before closing the connection.
         void Stop();
 
     private:
@@ -268,12 +269,17 @@ namespace Crafter {
     //     via OnStream()).
     //   - Unreliable, unordered datagrams (SendDatagram() / OnDatagram()).
     //
-    // Lifetime: ~ClientQUIC shuts the connection down and closes it, waiting
+    // Lifetime: ~ClientQUIC first lets the streams already stopped finish
-    // (bounded — 5s) for every stream on it to finish closing first, since
+    // (bounded — 1s, about a round trip when the peer answers): what was
-    // QUICStream::Stop only initiates a shutdown that completes asynchronously,
+    // sent on them, the FIN last, reaches the peer, so a server may reply
-    // and then (bounded — 1s, about a round trip when the peer answers) for
+    // with SendAsync(..., finish = true), let the stream go and delete the
-    // the shutdown to complete, so the peer is sent the close rather than
+    // connection at once. Then it shuts the connection down, which aborts
-    // left to time the connection out.
+    // any stream still open, 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, and then (bounded
+    // — 1s, about a round trip when the peer answers) for the shutdown to
+    // complete, so the peer is sent the close rather than left to time the
+    // connection out.
     // Streams outliving the connection are therefore safe to hold, though they
     // are dead once it goes; prefer destroying or moving them out first.
     //

project.cpp

@@ -195,6 +195,7 @@ extern "C" Configuration CrafterBuildProject(std::span<const std::string_view> a
             cfg.AddTest("ShouldBindLoopbackQUIC").Dependencies({ &cfg });
             cfg.AddTest("ShouldEchoWebTransport").Dependencies({ &cfg });
             cfg.AddTest("ShouldFallbackUnknownRoutes").Dependencies({ &cfg });
+            cfg.AddTest("ShouldFinishStoppedStreamsQUIC").Dependencies({ &cfg });
             cfg.AddTest("ShouldHonourQUICSettings").Dependencies({ &cfg });
             cfg.AddTest("ShouldNotDropEarlyStreams").Dependencies({ &cfg });
             cfg.AddTest("ShouldNotifyClosedConnectionQUIC").Dependencies({ &cfg });