//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); }