test(transport): drive a listening session over loopback

ShouldServeOverQuic runs a Session that listens for real and reaches it
with the client's own transport. It covers:
- the discovery record, and its removal when the Session ends;
- status, call, a mistyped command, get and set, and log --follow
  streaming as entries are logged and ending on time;
- six clients at once, and eight streams at once on one connection;
- a wrong token, a request that isn't JSON, and one over 1 MiB, each
  answered with an error;
- ending the Session with a follower waiting, which is told "shutting
  down" and is not sent a backlog it asked none of;
- a client finding the port dead in milliseconds.

The shutdown reply went missing in 4 runs of 30 until Crafter.Network's
~ClientQUIC let stopped streams finish before closing; 50 of 50 pass
since.

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

project.cpp

@@ -232,6 +232,7 @@ extern "C" Configuration CrafterBuildProject(std::span<const std::string_view> a
         for (const std::string_view test : coreTests) cfg.AddTest(test).Dependencies({ &core });
         for (const std::string_view test : libraryTests) cfg.AddTest(test).Dependencies({ &cfg });
         cfg.AddTest("ShouldRunClientCommands").Dependencies({ &cfg, &cli });
+        cfg.AddTest("ShouldServeOverQuic").Dependencies({ &cfg, &cli });
     }
     return cfg;
 }

tests/ShouldServeOverQuic/main.cpp

@@ -0,0 +1,264 @@
+// SPDX-License-Identifier: LGPL-3.0-only
+// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
+
+// The QUIC transport end to end: a Session that listens for real, reached
+// over loopback by the client's own code, as crafter-debug reaches it. What
+// the in-process tests can't see:
+//  - discovery lists the instance with its port, token, protocol, source and
+//    client, and drops it once the Session is gone;
+//  - status, call, get and set work through the client, and log --follow
+//    streams lines as they are logged, then ends;
+//  - a wrong token is refused, a request that isn't JSON is answered, and
+//    one over 1 MiB is answered with an error instead of being buffered;
+//  - requests in parallel, on streams of one connection and on separate
+//    connections, each get their own reply;
+//  - destroying the Session answers a waiting follower and closes promptly
+//    with a client still connected, and a client then finds the port dead
+//    within seconds.
+#include <stdio.h>
+#include <stdlib.h>
+import std;
+import Crafter.Thread;
+import Crafter.Network;
+import Crafter.Debug;
+import Crafter.Debug.Cli;
+using namespace Crafter::Debug;
+namespace fs = std::filesystem;
+using namespace std::chrono_literals;
+
+namespace {
+    std::int32_t failures = 0;
+
+    void Check(bool ok, std::string_view what) {
+        std::println("{} {}", ok ? "PASS" : "FAIL", what);
+        if (!ok) ++failures;
+    }
+
+    bool Has(const std::string& text, std::string_view part) {
+        return text.find(part) != std::string::npos;
+    }
+
+    struct AddArgs {
+        Required<std::int32_t> a;
+        std::int32_t b = 0;
+    };
+
+    struct Ran {
+        std::int32_t code = 0;
+        std::string out;
+        std::string err;
+    };
+
+    // crafter-debug, with its real transport, against the discovery
+    // directory the test set.
+    Ran Run(std::vector<std::string_view> args) {
+        std::ostringstream out;
+        std::ostringstream err;
+        Cli::Environment environment;
+        environment.connect = Cli::ConnectQuic;
+        environment.out = &out;
+        environment.err = &err;
+        environment.haveLldb = [] { return false; };
+        const std::int32_t code = Cli::Run(args, environment);
+        return { code, out.str(), err.str() };
+    }
+
+    // A request sent by hand, on its own stream: what came back, and how
+    // the stream ended.
+    struct Received {
+        std::mutex mutex;
+        std::condition_variable changed;
+        std::string bytes;
+        bool ended = false;
+        bool clean = false;
+    };
+
+    struct Raw {
+        std::shared_ptr<Received> received = std::make_shared<Received>();
+        Crafter::QUICStream stream;
+
+        bool Wait(std::chrono::milliseconds limit) {
+            std::unique_lock lock(received->mutex);
+            return received->changed.wait_for(lock, limit, [&] { return received->ended; });
+        }
+
+        std::string Bytes() {
+            std::lock_guard lock(received->mutex);
+            return received->bytes;
+        }
+    };
+
+    std::unique_ptr<Raw> Send(Crafter::ClientQUIC& client, std::string request) {
+        auto raw = std::make_unique<Raw>();
+        raw->stream = client.OpenStream();
+        const std::shared_ptr<Received> received = raw->received;
+        raw->stream.OnRecieve(
+            [received](std::span<const char> data) {
+                {
+                    std::lock_guard lock(received->mutex);
+                    received->bytes.append(data.data(), data.size());
+                }
+                received->changed.notify_all();
+            },
+            [received](bool clean) {
+                {
+                    std::lock_guard lock(received->mutex);
+                    received->ended = true;
+                    received->clean = clean;
+                }
+                received->changed.notify_all();
+            });
+        raw->stream.SendAsync(request.data(), static_cast<std::uint32_t>(request.size()), true, {});
+        return raw;
+    }
+
+    std::string Request(std::string_view token, std::string_view method, std::string_view params = "{}") {
+        return std::format(R"({{"token":"{}","method":"{}","params":{},"timeoutMs":5000}})", token, method, params) + "\n";
+    }
+
+    Crafter::QUICClientCredentials Insecure() {
+        Crafter::QUICClientCredentials credentials;
+        credentials.insecureNoServerValidation = true;
+        return credentials;
+    }
+}
+
+int main() {
+    std::thread([] {
+        std::this_thread::sleep_for(50s);
+        std::println("FAIL the test hung");
+        std::fflush(stdout);
+        std::_Exit(1);
+    }).detach();
+
+    const fs::path dir = fs::temp_directory_path() / "crafter-debug-ShouldServeOverQuic";
+    fs::remove_all(dir);
+    fs::create_directories(dir);
+    ::setenv("CRAFTER_DEBUG_DIR", dir.c_str(), 1);
+
+    Crafter::ThreadPool::Start(4);
+
+    Command("Math_Add", "Add two numbers.", [](const AddArgs& args) { return *args.a + args.b; });
+    float speed = 1.5f;
+    Watch("speed", speed);
+
+    auto session = std::make_unique<Session>(SessionOptions{ .name = "quic-test" });
+
+    std::atomic<bool> pumping = true;
+    std::thread app([&] {
+        std::uint32_t frame = 0;
+        while (pumping) {
+            Pump();
+            if (++frame % 10 == 0) Log("tick", "tick {}", frame);
+            std::this_thread::sleep_for(2ms);
+        }
+    });
+
+    // ---- discovery ----
+    const std::vector<Instance> live = ListInstances();
+    Check(live.size() == 1, "discovery lists the listening session");
+    if (live.size() != 1) {
+        pumping = false;
+        app.join();
+        std::fflush(stdout);
+        std::_Exit(1);
+    }
+    const InstanceRecord record = live.front().record;
+    Check(record.name == "quic-test" && record.port != 0 && record.token == session->Token(), "the record names it, its port and its token");
+    Check(record.protocol == protocolVersion && record.sourceId == Detail::SourceId() && record.client == Detail::ClientPath(),
+          "the record carries the protocol, source id and client path");
+
+    // ---- the client, over QUIC ----
+    Ran r = Run({ "status" });
+    Check(r.code == 0 && Has(r.out, "quic-test  frame"), std::format("status ({})", r.out + r.err));
+    r = Run({ "call", "Math_Add", "a=2", "b=3" });
+    Check(r.code == 0 && Has(r.out, "5"), std::format("call ({})", r.out + r.err));
+    r = Run({ "call", "Math_Ad", "a=2" });
+    Check(r.code == 6 && Has(r.err, "Math_Add"), std::format("an unknown command suggests the right one ({})", r.err));
+    r = Run({ "get", "speed" });
+    Check(r.code == 0 && Has(r.out, "1.5"), std::format("get ({})", r.out + r.err));
+    r = Run({ "set", "speed", "4" });
+    Check(r.code == 0, std::format("set ({})", r.out + r.err));
+    r = Run({ "get", "speed" });
+    Check(r.code == 0 && Has(r.out, "4"), std::format("get after set ({})", r.out + r.err));
+
+    const auto followStart = std::chrono::steady_clock::now();
+    r = Run({ "log", "--follow", "--for", "1s", "--cat", "tick" });
+    const auto followTook = std::chrono::steady_clock::now() - followStart;
+    Check(r.code == 0 && Has(r.out, "tick") && Has(r.err, "(follow ended)"), std::format("log --follow streams, then ends ({})", r.err));
+    Check(followTook >= 900ms && followTook < 5s, std::format("the follow lasted its --for ({} ms)", std::chrono::duration_cast<std::chrono::milliseconds>(followTook).count()));
+
+    // ---- in parallel, on separate connections ----
+    {
+        std::vector<std::thread> callers;
+        std::vector<Ran> results(6);
+        for (std::size_t i = 0; i < results.size(); ++i) {
+            callers.emplace_back([&results, i] {
+                const std::string a = std::format("a={}", 100 * (i + 1));
+                results[i] = Run({ "call", "Math_Add", a, "b=1" });
+            });
+        }
+        for (std::thread& caller : callers) caller.join();
+        bool all = true;
+        for (std::size_t i = 0; i < results.size(); ++i) {
+            all = all && results[i].code == 0 && Has(results[i].out, std::format("{}", 100 * (i + 1) + 1));
+        }
+        Check(all, "six clients at once each get their own answer");
+    }
+
+    // ---- by hand: what the client never sends ----
+    auto client = std::make_unique<Crafter::ClientQUIC>(std::string("127.0.0.1"), record.port, std::string(alpn), Insecure());
+    {
+        std::unique_ptr<Raw> wrong = Send(*client, Request("not-the-token", "debug.status"));
+        Check(wrong->Wait(5s) && Has(wrong->Bytes(), "Unauthorized"), std::format("a wrong token is refused ({})", wrong->Bytes()));
+
+        std::unique_ptr<Raw> garbage = Send(*client, "this is not json\n");
+        Check(garbage->Wait(5s) && Has(garbage->Bytes(), "InvalidRequest"), std::format("a request that isn't JSON is answered ({})", garbage->Bytes()));
+
+        std::unique_ptr<Raw> huge = Send(*client, std::string(std::size_t{ 2 } << 20, ' '));
+        Check(huge->Wait(10s) && Has(huge->Bytes(), "InvalidRequest") && Has(huge->Bytes(), "at most"),
+              std::format("a request over 1 MiB is answered with an error ({})", huge->Bytes().substr(0, 200)));
+
+        std::vector<std::unique_ptr<Raw>> parallel;
+        for (std::int32_t i = 0; i < 8; ++i) {
+            parallel.push_back(Send(*client, Request(record.token, "Math_Add", std::format(R"({{"a":{},"b":1000}})", i))));
+        }
+        bool all = true;
+        for (std::int32_t i = 0; i < 8; ++i) {
+            all = all && parallel[i]->Wait(5s) && Has(parallel[i]->Bytes(), std::format(R"({{"result":{}}})", 1000 + i));
+        }
+        Check(all, "eight streams of one connection at once each get their own answer");
+    }
+
+    // ---- the end, with a follower waiting ----
+    std::unique_ptr<Raw> follower = Send(*client, Request(record.token, "debug.log", R"({"follow":true,"forMs":60000,"last":0})"));
+    bool following = false;
+    for (std::int32_t attempt = 0; attempt < 100 && !following; ++attempt) {
+        std::unique_ptr<Raw> status = Send(*client, Request(record.token, "debug.status"));
+        following = status->Wait(5s) && Has(status->Bytes(), R"("following":1)");
+        if (!following) std::this_thread::sleep_for(20ms);
+    }
+    Check(following, "the follower is registered");
+    pumping = false;
+    app.join();
+
+    const auto endStart = std::chrono::steady_clock::now();
+    session.reset();
+    const auto endTook = std::chrono::steady_clock::now() - endStart;
+    Check(endTook < 3s, std::format("the Session ends promptly with a client connected ({} ms)", std::chrono::duration_cast<std::chrono::milliseconds>(endTook).count()));
+    Check(follower->Wait(5s) && Has(follower->Bytes(), "ShuttingDown") && !Has(follower->Bytes(), "Math_Add"),
+          std::format("the follower, with no backlog, is told the app is shutting down ({})", follower->Bytes().substr(0, 400)));
+    Check(ListInstances().empty(), "discovery forgets the session");
+    follower.reset();
+    client.reset();
+
+    const auto deadStart = std::chrono::steady_clock::now();
+    const std::unique_ptr<Cli::Connection> dead = Cli::ConnectQuic(record);
+    const auto reply = dead->Call(Value::Parse(Request(record.token, "debug.status")), 5s, {});
+    const auto deadTook = std::chrono::steady_clock::now() - deadStart;
+    Check(!reply && reply.error().kind == Cli::CallError::Kind::Unreachable && deadTook < 5s,
+          std::format("a client finds the port dead in {} ms", std::chrono::duration_cast<std::chrono::milliseconds>(deadTook).count()));
+
+    std::fflush(stdout);
+    std::_Exit(failures == 0 ? 0 : 1);
+}