From bdca27b871674c85d36a7ad1a71798fd0800cb9e Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sun, 1 Nov 2026 02:32:48 +0100 Subject: [PATCH] test(quic): receive by callback, one callback at a time, ended once ShouldRecieveByCallbackQUIC, seventeen checks. Bytes and a FIN that came before OnRecieve are delivered by it before it returns. Registered while 1000 KiB keep arriving, with the first callbacks slowed to widen the window and an echo sent from inside onData, every byte arrives in order and no two of the stream's callbacks, onSent included, overlap; the echo comes back whole. With every pool worker parked, 256 KiB goes out and comes back by callback and SendAsync alone. The peer's Abort() and a lost connection each end the stream not clean, once. Abort() from inside onData caps a 1 MiB upload: no onData after it, onEnd(false), and the peer's receive side ends while its next send is refused. A wrapper destroyed inside its own onEnd, or moved while bytes flow, breaks nothing; one destroyed on another thread mid-flow has no callback run once its destructor returned. A second OnRecieve throws, and a stream that cannot receive, or none, gets onEnd(false) at once. With the serialisation disabled the overlap and order checks fail every run; with Unregister's wait removed the destroyed-wrapper check fails in most. Co-Authored-By: Claude Opus 5.5 --- project.cpp | 1 + tests/ShouldRecieveByCallbackQUIC/main.cpp | 526 +++++++++++++++++++++ 2 files changed, 527 insertions(+) create mode 100644 tests/ShouldRecieveByCallbackQUIC/main.cpp diff --git a/project.cpp b/project.cpp index e23efb9..74fb987 100644 --- a/project.cpp +++ b/project.cpp @@ -206,6 +206,7 @@ extern "C" Configuration CrafterBuildProject(std::span a cfg.AddTest("ShouldSendRecieveQUICStream").Dependencies({ &cfg }); cfg.AddTest("ShouldSendWithoutBlockingQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldReportClosedQUICStream").Dependencies({ &cfg }); + cfg.AddTest("ShouldRecieveByCallbackQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldReportListenerPortQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldReturnFromSendSyncQUIC").Dependencies({ &cfg }); cfg.AddTest("ShouldShareSelfSignedCertificateQUIC").Dependencies({ &cfg }); diff --git a/tests/ShouldRecieveByCallbackQUIC/main.cpp b/tests/ShouldRecieveByCallbackQUIC/main.cpp new file mode 100644 index 0000000..e2194b6 --- /dev/null +++ b/tests/ShouldRecieveByCallbackQUIC/main.cpp @@ -0,0 +1,526 @@ +//SPDX-License-Identifier: LGPL-3.0-only +//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// QUICStream::OnRecieve delivers a stream's bytes and its end by callback, +// holding no thread while nothing arrives. Until now every way to wait for +// bytes held one: RecieveSync and RecieveUntilCloseSync block their caller +// with no bound, and RecieveAsync and RecieveUntilCloseAsync block a worker +// of the pool the host shares, which for a stream a client keeps open (a log +// follower) is for good. A server embedded in an app cannot afford either. +// +// - Bytes that arrived before OnRecieve are delivered by it, before it +// returns, and the end with them: clean for a FIN. +// - Registered while bytes keep arriving, with a callback slow enough to +// let the registering thread and msquic's worker overlap, and an echo +// sent from inside onData: no two callbacks of the stream ever run at +// once, and every byte arrives in order. +// - With every pool worker parked, a stream is echoed both ways, callbacks +// and SendAsync only: nothing needed the pool. +// - The end fires once: not clean for the peer's Abort() and for a lost +// connection. +// - Abort() from inside onData enforces a byte cap: no onData after it, +// onEnd(false) once, and the peer sees both directions reset. +// - The wrapper destroyed inside its own onEnd, or moved while bytes flow, +// breaks nothing; destroyed from another thread mid-flow, no callback +// runs once its destructor has returned. +// - A second OnRecieve throws; a stream that cannot receive, or no stream, +// gets onEnd(false) at once. + +#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; +} + +bool WaitFor(std::function done, std::chrono::milliseconds limit) { + const auto deadline = std::chrono::steady_clock::now() + limit; + while (!done()) { + if (std::chrono::steady_clock::now() >= deadline) return false; + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } + return true; +} + +char Pattern(std::uint64_t offset) { return static_cast((offset * 131 + 7) & 0xff); } + +bool FollowsPattern(const std::string& bytes, std::size_t expected) { + if (bytes.size() != expected) return false; + for (std::size_t i = 0; i < bytes.size(); ++i) { + if (bytes[i] != Pattern(i)) return false; + } + return true; +} + +const std::string alpn = "crafter/test-on-recieve"; + +QUICClientCredentials ClientCredentials() { + QUICClientCredentials creds; + creds.insecureNoServerValidation = true; + return creds; +} + +// What one stream's callbacks saw. Shared with them, and leaked, so a +// callback that comes late on a failure path finds it alive. +struct Seen { + std::mutex mutex; + std::string bytes; + int ends = 0; + bool clean = false; + int dataAfterEnd = 0; + std::atomic inside = 0; // callbacks of this stream running now + std::atomic overlaps = 0; // times two ran at once + + void Enter() { if (inside.fetch_add(1) != 0) ++overlaps; } + void Leave() { --inside; } + + std::function)> OnData() { + return [this](std::span data) { + Enter(); + { + std::lock_guard lock(mutex); + if (ends != 0) ++dataAfterEnd; + bytes.append(data.begin(), data.end()); + } + Leave(); + }; + } + std::function OnEnd() { + return [this](bool wasClean) { + Enter(); + { + std::lock_guard lock(mutex); + ++ends; + clean = wasClean; + } + Leave(); + }; + } + int Ends() { + std::lock_guard lock(mutex); + return ends; + } + std::string Bytes() { + std::lock_guard lock(mutex); + return bytes; + } +}; + +// Every pool worker parked until Release(), which returns once each has +// left, so nothing reads this after its owner is gone. +struct Parking { + std::atomic release = false; + std::atomic parked = 0; + std::atomic left = 0; + unsigned tasks = 0; + + bool Park() { + const unsigned workers = ThreadPool::activeThreads.load(); + tasks = workers * 2; + for (unsigned i = 0; i < tasks; ++i) { + ThreadPool::Enqueue([this] { + ++parked; + while (!release.load()) std::this_thread::sleep_for(std::chrono::milliseconds(1)); + ++left; + }); + } + return WaitFor([&] { return parked.load() >= workers; }, std::chrono::seconds(10)); + } + void Release() { + release = true; + WaitFor([&] { return left.load() == tasks; }, std::chrono::seconds(10)); + } +}; + +// A listener whose every accepted stream goes to `onStream`, as ListenerQUIC +// delivers it: on a pool task. Wrappers the handler keeps go in `kept`. +struct Server { + std::mutex mutex; + std::vector> kept; + std::unique_ptr listener; + + explicit Server(std::function)> onStream) { + QUICServerCredentials creds; + creds.selfSigned = true; + listener = std::make_unique("127.0.0.1", 0, alpn, creds, [this, onStream](ClientQUIC* peer) { + peer->OnStream([this, onStream](QUICStream stream) { + onStream(*this, std::make_shared(std::move(stream))); + }); + }); + listener->ListenAsyncAsync(); + } + void Keep(std::shared_ptr stream) { + std::lock_guard lock(mutex); + kept.push_back(std::move(stream)); + } + std::uint16_t Port() const { return listener->Port(); } +}; + +} // namespace + +int main() { + ThreadPool::Start(); + std::thread([] { + std::this_thread::sleep_for(std::chrono::seconds(50)); + std::println("FAIL timed out"); + std::fflush(stdout); + std::_Exit(1); + }).detach(); + + try { + // ── What arrived first is delivered by OnRecieve ───────────────────── + { + Seen& seen = *new Seen; + std::atomic registering = false; + std::atomic deliveredInside = 0; + Server& server = *new Server([&](Server& self, std::shared_ptr stream) { + // Long enough for "hello" and its FIN to be waiting. + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + registering = true; + auto onData = seen.OnData(); + auto onEnd = seen.OnEnd(); + stream->OnRecieve( + [&, onData](std::span data) { if (registering) ++deliveredInside; onData(data); }, + [&, onEnd](bool clean) { if (registering) ++deliveredInside; onEnd(clean); }); + registering = false; + self.Keep(stream); + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + stream.SendSync("hello", 5, true); + Check(WaitFor([&] { return seen.Ends() == 1; }, std::chrono::seconds(10)) && seen.Bytes() == "hello" && seen.clean, + std::format("bytes and a FIN that came before OnRecieve are delivered by it: '{}', clean", seen.Bytes())); + Check(deliveredInside.load() == 2, "on the registering thread, before OnRecieve returned"); + } + + // ── Registered mid-flow: one callback at a time, every byte in order ── + { + Seen& seen = *new Seen; + auto slow = std::make_shared>(50); + Server& server = *new Server([&seen, slow](Server& self, std::shared_ptr stream) { + // Bytes keep coming meanwhile, so the backlog this thread + // delivers and the worker's new arrivals overlap in time. + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + auto onData = seen.OnData(); + std::weak_ptr weak = stream; + stream->OnRecieve( + [&seen, slow, onData, weak](std::span data) { + if (slow->fetch_sub(1) > 0) std::this_thread::sleep_for(std::chrono::microseconds(200)); + onData(data); + // Echoed from inside the callback; its onSent is one + // of the stream's callbacks too. + if (auto s = weak.lock()) { + s->SendAsync(data.data(), static_cast(data.size()), false, [&seen](bool) { + seen.Enter(); + seen.Leave(); + }); + } + }, + [weak, onEnd = seen.OnEnd()](bool clean) { + onEnd(clean); + if (auto s = weak.lock()) s->SendAsync(nullptr, 0, true, nullptr); + }); + self.Keep(stream); + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + Seen& echo = *new Seen; + stream.OnRecieve(echo.OnData(), echo.OnEnd()); + constexpr std::size_t total = 2000 * 512; + std::vector chunk(512); + for (std::size_t sent = 0; sent < total; sent += chunk.size()) { + for (std::size_t i = 0; i < chunk.size(); ++i) chunk[i] = Pattern(sent + i); + stream.SendAsync(chunk.data(), static_cast(chunk.size()), sent + chunk.size() == total, nullptr); + } + const bool ended = WaitFor([&] { return seen.Ends() == 1 && echo.Ends() == 1; }, std::chrono::seconds(20)); + Check(ended && seen.clean && FollowsPattern(seen.Bytes(), total) && seen.dataAfterEnd == 0, + std::format("registered mid-flow, every byte arrived in order, then a clean end ({} of {})", seen.Bytes().size(), total)); + Check(seen.overlaps.load() == 0, "and no two of the stream's callbacks, onSent included, ever ran at once"); + Check(echo.clean && FollowsPattern(echo.Bytes(), total), "the echo sent from inside onData came back whole"); + } + + // ── No pool ────────────────────────────────────────────────────────── + { + Seen& seen = *new Seen; + std::atomic registered = false; + Server& server = *new Server([&](Server& self, std::shared_ptr stream) { + std::weak_ptr weak = stream; + stream->OnRecieve( + [&, weak](std::span data) { + seen.OnData()(data); + if (auto s = weak.lock()) s->SendAsync(data.data(), static_cast(data.size()), false, nullptr); + }, + [weak](bool) { if (auto s = weak.lock()) s->SendAsync(nullptr, 0, true, nullptr); }); + self.Keep(stream); + registered = true; + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + Seen& echo = *new Seen; + stream.OnRecieve(echo.OnData(), echo.OnEnd()); + const char first = Pattern(0); + stream.SendAsync(&first, 1, false, nullptr); + // The server's OnStream is a pool task: let it run first. + WaitFor([&] { return registered.load(); }, std::chrono::seconds(10)); + + Parking& parking = *new Parking; + const bool allParked = parking.Park(); + constexpr std::size_t total = 256 * 1024; + std::vector rest(total - 1); + for (std::size_t i = 0; i < rest.size(); ++i) rest[i] = Pattern(i + 1); + stream.SendAsync(rest.data(), static_cast(rest.size()), true, nullptr); + const bool ended = WaitFor([&] { return echo.Ends() == 1; }, std::chrono::seconds(10)); + Check(allParked && ended && echo.clean && FollowsPattern(echo.Bytes(), total), + std::format("with all {} pool workers parked, 256 KiB went out and came back by callback ({} bytes)", + parking.parked.load(), echo.Bytes().size())); + parking.Release(); + } + + // ── Unclean ends ───────────────────────────────────────────────────── + { + std::mutex mutex; + std::vector seens; + Server& server = *new Server([&](Server& self, std::shared_ptr stream) { + Seen& seen = *new Seen; + { + std::lock_guard lock(mutex); + seens.push_back(&seen); + } + stream->OnRecieve(seen.OnData(), seen.OnEnd()); + self.Keep(stream); + }); + auto latest = [&]() -> Seen* { + std::lock_guard lock(mutex); + return seens.empty() ? nullptr : seens.back(); + }; + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + { + QUICStream stream = client.OpenStream(); + stream.SendSync("x", 1, false); + WaitFor([&] { return latest() != nullptr && latest()->Bytes() == "x"; }, std::chrono::seconds(10)); + stream.Abort(7); + Seen* seen = latest(); + Check(seen && WaitFor([&] { return seen->Ends() == 1; }, std::chrono::seconds(10)) && !seen->clean, + "the peer's Abort() ends the stream, not clean"); + } + { + ClientQUIC lost("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = lost.OpenStream(); + Seen* before = latest(); + stream.SendSync("y", 1, false); + WaitFor([&] { return latest() != before && latest()->Bytes() == "y"; }, std::chrono::seconds(10)); + Seen* seen = latest(); + lost.Stop(); // the connection goes; the stream gets no FIN + Check(seen != before && WaitFor([&] { return seen->Ends() == 1; }, std::chrono::seconds(10)) && !seen->clean, + "a lost connection ends it, not clean"); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + Check(seen->Ends() == 1, "each end exactly once"); + } + } + + // ── A byte cap, by Abort() from inside onData ──────────────────────── + { + Seen& seen = *new Seen; + std::atomic afterAbort = 0; + std::atomic aborted = false; + constexpr std::size_t cap = 64 * 1024; + Server& server = *new Server([&](Server& self, std::shared_ptr stream) { + std::weak_ptr weak = stream; + stream->OnRecieve( + [&, weak](std::span data) { + if (aborted) ++afterAbort; + seen.OnData()(data); + if (seen.Bytes().size() > cap && !aborted) { + aborted = true; + if (auto s = weak.lock()) s->Abort(0x42); + } + }, + seen.OnEnd()); + self.Keep(stream); + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + Seen& back = *new Seen; + stream.OnRecieve(back.OnData(), back.OnEnd()); + constexpr int sends = 16; + std::atomic reported = 0; + std::atomic refused = 0; + const std::vector chunk(64 * 1024, 'z'); + for (int i = 0; i < sends; ++i) { + stream.SendAsync(chunk.data(), static_cast(chunk.size()), false, [&](bool sent) { + if (!sent) ++refused; + ++reported; + }); + } + const bool ended = WaitFor([&] { return seen.Ends() == 1; }, std::chrono::seconds(10)); + Check(ended && !seen.clean && afterAbort.load() == 0 && seen.Bytes().size() > cap + && seen.Bytes().size() < sends * chunk.size(), + std::format("Abort() inside onData stopped the stream at {} KiB: no onData after it, onEnd(false)", + seen.Bytes().size() / 1024)); + // Its sends may all report true: msquic takes them into its send + // buffer before the reset arrives, and true means just that. A + // send made once the reset has arrived is the one refused. + const bool backEnded = WaitFor([&] { return back.Ends() == 1 && reported.load() == sends; }, std::chrono::seconds(10)); + bool probeRefused = false; + for (int attempt = 0; attempt < 100 && !probeRefused; ++attempt) { + auto result = std::make_shared>(-1); + stream.SendAsync("p", 1, false, [result](bool sent) { *result = sent ? 1 : 0; }); + WaitFor([&] { return result->load() != -1; }, std::chrono::seconds(5)); + probeRefused = result->load() == 0; + if (!probeRefused) std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } + Check(backEnded && !back.clean && probeRefused, + std::format("the peer saw both directions reset: its receive side ended not clean, and its sends are refused ({} of the first {} already were)", + refused.load(), sends)); + } + + // ── The wrapper destroyed or moved under its callbacks ─────────────── + { + // Destroyed inside its own onEnd: the callback holds the last + // reference. + std::atomic destroyedInside = 0; + std::atomic streams = 0; + Server& server = *new Server([&](Server&, std::shared_ptr stream) { + ++streams; + auto holder = std::make_shared>(stream); + stream->OnRecieve([](std::span) {}, [holder, &destroyedInside](bool) { + holder->reset(); + ++destroyedInside; + }); + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + for (int i = 0; i < 3; ++i) { + QUICStream stream = client.OpenStream(); + stream.SendSync("bye", 3, true); + } + Check(WaitFor([&] { return destroyedInside.load() == 3; }, std::chrono::seconds(10)), + "a wrapper destroyed inside its own onEnd breaks nothing, three times over"); + } + { + // Moved while bytes flow: the callbacks never touch the wrapper. + Seen& seen = *new Seen; + std::atomic registered = false; + std::shared_ptr first; + Server& server = *new Server([&](Server&, std::shared_ptr stream) { + stream->OnRecieve(seen.OnData(), seen.OnEnd()); + first = stream; + registered = true; + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + const char lead = Pattern(0); + stream.SendSync(&lead, 1, false); + WaitFor([&] { return registered.load(); }, std::chrono::seconds(10)); + constexpr std::size_t total = 512 * 1024; + std::thread sender([&] { + std::vector chunk(1024); + for (std::size_t sent = 1; sent < total; sent += chunk.size()) { + const std::size_t size = std::min(chunk.size(), total - sent); + for (std::size_t i = 0; i < size; ++i) chunk[i] = Pattern(sent + i); + stream.SendSync(chunk.data(), static_cast(size), sent + size == total); + } + }); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + auto second = std::make_shared(std::move(*first)); + first.reset(); + sender.join(); + Check(WaitFor([&] { return seen.Ends() == 1; }, std::chrono::seconds(10)) && seen.clean + && FollowsPattern(seen.Bytes(), total), + "a wrapper moved while bytes flow keeps receiving, every byte in order"); + second.reset(); + } + { + // Destroyed from another thread mid-flow: no callback after. + // What the callbacks touch is leaked, in case they do run late. + struct Watch { + Seen seen; + std::atomic destroyed = false; + std::atomic afterDestroyed = 0; + std::shared_ptr held; + std::atomic registered = false; + }; + Watch& watch = *new Watch; + Seen& seen = watch.seen; + std::atomic& destroyed = watch.destroyed; + std::atomic& afterDestroyed = watch.afterDestroyed; + std::shared_ptr& held = watch.held; + std::atomic& registered = watch.registered; + Server& server = *new Server([&watch](Server&, std::shared_ptr stream) { + stream->OnRecieve( + [&watch](std::span) { + watch.seen.Enter(); + // Slow, so destroying the wrapper often catches one. + std::this_thread::sleep_for(std::chrono::microseconds(500)); + if (watch.destroyed) ++watch.afterDestroyed; + watch.seen.Leave(); + }, + [&watch](bool) { if (watch.destroyed) ++watch.afterDestroyed; }); + watch.held = stream; + watch.registered = true; + }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + stream.SendSync("go", 2, false); + WaitFor([&] { return registered.load(); }, std::chrono::seconds(10)); + std::atomic stop = false; + std::thread sender([&] { + const std::vector chunk(256, 'q'); + while (!stop) { + stream.SendAsync(chunk.data(), static_cast(chunk.size()), false, nullptr); + std::this_thread::sleep_for(std::chrono::microseconds(200)); + } + }); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + held.reset(); // ~QUICStream on this thread, callbacks running on the worker + destroyed = true; + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + stop = true; + sender.join(); + Check(afterDestroyed.load() == 0 && seen.overlaps.load() == 0, + "once a wrapper destroyed on another thread has gone, no callback of it runs again"); + } + + // ── Misuse and edge cases ──────────────────────────────────────────── + { + Server& server = *new Server([](Server& self, std::shared_ptr stream) { self.Keep(stream); }); + ClientQUIC client("127.0.0.1", server.Port(), alpn, ClientCredentials()); + QUICStream stream = client.OpenStream(); + stream.OnRecieve([](std::span) {}, [](bool) {}); + bool threw = false; + try { + stream.OnRecieve([](std::span) {}, [](bool) {}); + } catch (const QUICException&) { + threw = true; + } + Check(threw, "a second OnRecieve on one stream throws QUICException"); + + QUICStream sendOnly = client.OpenStream(/*unidirectional=*/true); + std::atomic sendOnlyEnds = 0; + std::atomic sendOnlyClean = true; + sendOnly.OnRecieve([](std::span) {}, [&](bool clean) { ++sendOnlyEnds; sendOnlyClean = clean; }); + Check(sendOnlyEnds.load() == 1 && !sendOnlyClean.load(), "a stream that cannot receive gets onEnd(false) at once"); + + QUICStream none; + std::atomic noneEnds = 0; + none.OnRecieve([](std::span) {}, [&](bool clean) { if (!clean) ++noneEnds; }); + Check(noneEnds.load() == 1, "so does no stream at all"); + } + } catch (const std::exception& e) { + Check(false, std::format("threw: {}", e.what())); + } + + std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed"); + // As the other QUIC tests: skip the static teardown, which waits for the + // accepted peers' idle timeout. _Exit doesn't flush. + std::fflush(stdout); + std::_Exit(failures == 0 ? 0 : 1); +}