From 4876ff14a80fe5ef729ce64e59cb3bf30b89536d Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sun, 1 Nov 2026 03:03:20 +0100 Subject: [PATCH] test(quic): a reply sent just before the connection is deleted arrives Twenty connections in turn: the client sends a request, and the server answers 64 KiB with SendAsync(finish = true), destroys the stream wrapper and deletes its ClientQUIC at once, as a server shutting down does. Each answer must arrive whole with a clean end, and each delete must stay well under the 1 s bound. Before the fix every answer was reset. Co-Authored-By: Claude Opus 5.5 --- tests/ShouldFinishStoppedStreamsQUIC/main.cpp | 153 ++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 tests/ShouldFinishStoppedStreamsQUIC/main.cpp diff --git a/tests/ShouldFinishStoppedStreamsQUIC/main.cpp b/tests/ShouldFinishStoppedStreamsQUIC/main.cpp new file mode 100644 index 0000000..2fb5c49 --- /dev/null +++ b/tests/ShouldFinishStoppedStreamsQUIC/main.cpp @@ -0,0 +1,153 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// ~ClientQUIC lets the streams its owner already stopped finish before it +// closes the connection. It used to shut the connection down at once, and +// msquic aborts every stream with the connection, along with whatever of +// theirs it had not sent yet: a server that answered with +// SendAsync(..., finish = true), let the stream go and deleted the +// connection lost the answer whenever the shutdown, queued right behind the +// send, reached msquic's worker before the flush that would have sent it. +// Crafter.Debug's session, which answers every request still waiting when +// it ends and then closes its connections, lost that answer in 4 runs of 30. +// +// Twenty connections in turn. On each, the client sends a request; the +// server answers 64 KiB with SendAsync(finish = true), destroys the stream +// wrapper and deletes its ClientQUIC at once, as such a server does. The +// client must get every byte and a clean end each time, and the deletes +// must stay quick: a stopped stream finishes about a round trip after its +// last send, well inside the 1 s bound. + +#include + +import Crafter.Network; +import Crafter.Thread; +import std; +using namespace Crafter; + +namespace { + +int failures = 0; + +void Check(bool ok, std::string_view what) { + std::println("{} {}", ok ? "PASS" : "FAIL", what); + if (!ok) ++failures; +} + +// What the client's stream received, and how it ended. +struct Seen { + std::mutex mutex; + std::condition_variable changed; + std::size_t bytes = 0; + bool ended = false; + bool clean = false; +}; + +// The server's side of the connection being tested, handed over by the +// accept thread and the pool task that delivers the stream. +struct Accepted { + std::mutex mutex; + std::condition_variable changed; + ClientQUIC* peer = nullptr; + std::optional stream; +}; + +} // namespace + +int main() { + ThreadPool::Start(); + std::thread([] { + std::this_thread::sleep_for(std::chrono::seconds(45)); + std::println("FAIL timed out"); + std::fflush(stdout); + std::_Exit(1); + }).detach(); + + const std::string alpn = "crafter/test-finish-stopped"; + QUICServerCredentials serverCreds; + serverCreds.selfSigned = true; + QUICClientCredentials clientCreds; + clientCreds.insecureNoServerValidation = true; + + constexpr std::size_t answerSize = 64 * 1024; + const std::string answer(answerSize, 'a'); + auto accepted = std::make_shared(); + + try { + ListenerQUIC listener("127.0.0.1", 0, alpn, serverCreds, [accepted](ClientQUIC* peer) { + peer->OnStream([accepted](QUICStream stream) { + { + std::lock_guard lock(accepted->mutex); + accepted->stream.emplace(std::move(stream)); + } + accepted->changed.notify_all(); + }); + { + std::lock_guard lock(accepted->mutex); + accepted->peer = peer; + } + accepted->changed.notify_all(); + }); + listener.ListenAsyncSync(); + + int whole = 0; + std::chrono::milliseconds slowestDelete{ 0 }; + for (int i = 0; i < 20; ++i) { + ClientQUIC client("127.0.0.1", listener.Port(), alpn, clientCreds); + auto seen = std::make_shared(); + QUICStream request = client.OpenStream(); + request.OnRecieve( + [seen](std::span data) { + std::lock_guard lock(seen->mutex); + seen->bytes += data.size(); + }, + [seen](bool clean) { + { + std::lock_guard lock(seen->mutex); + seen->ended = true; + seen->clean = clean; + } + seen->changed.notify_all(); + }); + request.SendSync("ask", 3, true); + + ClientQUIC* peer = nullptr; + std::optional stream; + { + std::unique_lock lock(accepted->mutex); + if (!accepted->changed.wait_for(lock, std::chrono::seconds(5), [&] { return accepted->peer != nullptr && accepted->stream.has_value(); })) { + std::println(" connection {}: the server never saw the stream", i); + break; + } + peer = std::exchange(accepted->peer, nullptr); + stream = std::move(accepted->stream); + accepted->stream.reset(); + } + + // The server's last word, and then the end of it all at once. + stream->SendAsync(answer.data(), static_cast(answer.size()), true, {}); + stream.reset(); + const auto deleting = std::chrono::steady_clock::now(); + delete peer; + slowestDelete = std::max(slowestDelete, std::chrono::duration_cast(std::chrono::steady_clock::now() - deleting)); + + std::unique_lock lock(seen->mutex); + seen->changed.wait_for(lock, std::chrono::seconds(5), [&] { return seen->ended; }); + if (seen->ended && seen->clean && seen->bytes == answerSize) { + ++whole; + } else { + std::println(" connection {}: {} of {} bytes, {}", i, seen->bytes, answerSize, + !seen->ended ? "no end" : seen->clean ? "a clean end" : "reset"); + } + } + Check(whole == 20, std::format("each of 20 answers arrived whole before its connection closed ({} did)", whole)); + Check(slowestDelete < std::chrono::milliseconds(500), std::format("each delete waited only for its answer (slowest {} ms)", slowestDelete.count())); + } catch (const std::exception& e) { + Check(false, std::format("threw: {}", e.what())); + } + + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + // msquic's static teardown would wait on what the listener still holds. + std::fflush(stdout); + std::_Exit(failures == 0 ? 0 : 1); +}