From 47f2325664f1be9d852f1600cfbed722d14f688a Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sun, 1 Nov 2026 03:06:56 +0100 Subject: [PATCH] feat(transport): serve the session over QUIC on loopback A Session with listen = true (the default) now listens on 127.0.0.1 with the ALPN crafter-debug/1 and writes its discovery record (port, token, protocol, source id, client path), which goes first when it ends. One request per stream the client opens: the request and a FIN in, zero or more stream lines and then the reply and a FIN out, one JSON object per line. It holds no thread while nothing arrives. Crafter.Network delivers a new stream on the pool, its bytes on msquic's worker through OnRecieve, and replies leave through SendAsync, which never waits. The handler, which takes the session's mutex, runs on a pool task rather than inside the stream's callback, so no msquic worker ever waits on the session, and the session's threads only send, which Network allows under any lock. Stream wrappers are let go outside every lock their callbacks take. The rules head the implementation. Limits: a request over 1 MiB is answered with an error, not buffered, and a log follower more than 8 MiB behind is reset. quit --force exits 200 ms after its reply was taken, from a thread of its own, since "sent" means msquic has the bytes, not that they left. A client killed without closing is let go at the 10 s idle timeout. The Session refuses a pool without a worker: Network hands it new streams there, and with Start(1) nothing would run them. The partition exports nothing, so an app's view of the module is unchanged and it does not see Crafter.Network's types. Crafter.Network comes from forgejo, unpinned, with the args unchanged; --local takes ../Crafter.Network, and $CRAFTER_DEBUG_NETWORK any project.cpp, for the branch the callback API is on until it lands. Co-Authored-By: Claude Opus 5.5 --- implementations/Crafter.Debug-Session.cpp | 40 +++ implementations/Crafter.Debug-Transport.cpp | 296 ++++++++++++++++++++ interfaces/Crafter.Debug-Transport.cppm | 49 ++++ interfaces/Crafter.Debug.cppm | 3 + project.cpp | 31 +- 5 files changed, 416 insertions(+), 3 deletions(-) create mode 100644 implementations/Crafter.Debug-Transport.cpp create mode 100644 interfaces/Crafter.Debug-Transport.cppm diff --git a/implementations/Crafter.Debug-Session.cpp b/implementations/Crafter.Debug-Session.cpp index a3d4b35..d4df4b3 100644 --- a/implementations/Crafter.Debug-Session.cpp +++ b/implementations/Crafter.Debug-Session.cpp @@ -7,11 +7,13 @@ module; #include module Crafter.Debug:Session_impl; import std; +import Crafter.Thread; import Crafter.Debug.Core; import :Registry; import :Timeline; import :Condition; import :Session; +import :Transport; using namespace Crafter::Debug; using Crafter::Debug::Detail::Exchange; @@ -285,6 +287,11 @@ namespace Crafter::Debug { std::optional builtins; + // When listening: the transport, and the discovery record that + // tells clients where it is. + std::unique_ptr server; + std::optional registration; + explicit State(SessionOptions given) : options(std::move(given)) { name = options.name.empty() ? ProgramName() : options.name; } @@ -296,6 +303,30 @@ namespace Crafter::Debug { builtins->Watch("fps", [this] { return fps.load(); }, { .help = "Pumps per second, smoothed" }); } + // Last in the constructor: once the record is written, agents find + // this instance and send it requests. + void Listen(const std::shared_ptr& self) { + // Crafter.Network hands new streams to the pool, and this hands + // each request on; Start(1) gives the pool no worker to run them. + if (Crafter::ThreadPool::activeThreads.load() == 0) { + throw std::logic_error("Crafter.Debug: the Session needs Crafter.Thread's pool with a worker thread. " + "Call ThreadPool::Start() before creating it (Start(1) starts no worker)."); + } + const std::weak_ptr weak = self; + server = std::make_unique(options.port, [weak](const Value& request, std::shared_ptr exchange) { + if (const std::shared_ptr alive = weak.lock()) return alive->Handle(request, exchange); + exchange->Reply(MakeErrorReply(Codes::shuttingDown, "the app is shutting down")); + }); + registration.emplace(InstanceRecord{ + .name = name, + .port = server->Port(), + .token = token, + .protocol = protocolVersion, + .sourceId = std::string(Detail::SourceId()), + .client = std::string(Detail::ClientPath()), + }); + } + void Agent(std::string_view method, const Value& params) { const std::string text = params.IsNull() ? std::string(method) : std::format("{} {}", method, Abbreviated(params.Dump())); Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Agent, "agent", text, std::source_location::current())); @@ -1033,14 +1064,23 @@ Session::Session(SessionOptions options) : state(std::make_shared(std::mo // A debugger the agent starts is never this process's ancestor, // which Yama's ptrace_scope 1 would otherwise require. ::prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0); + if (state->options.listen) state->Listen(state); } catch (...) { + state->registration.reset(); + state->server.reset(); + state->Shutdown(); current.store(nullptr); throw; } } Session::~Session() { + // Out of the listing first, so no agent picks an instance on its way + // out; then every waiting request answered; then the connections + // closed, which lets those answers leave first. + state->registration.reset(); state->Shutdown(); + state->server.reset(); current.store(nullptr); } diff --git a/implementations/Crafter.Debug-Transport.cpp b/implementations/Crafter.Debug-Transport.cpp new file mode 100644 index 0000000..cc8f225 --- /dev/null +++ b/implementations/Crafter.Debug-Transport.cpp @@ -0,0 +1,296 @@ +// SPDX-License-Identifier: LGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module Crafter.Debug:Transport_impl; +import std; +import Crafter.Network; +import Crafter.Thread; +import Crafter.Debug.Core; +import :Session; +import :Transport; + +using namespace Crafter::Debug; +using Crafter::ClientQUIC; +using Crafter::ListenerQUIC; +using Crafter::QUICServerCredentials; +using Crafter::QUICStream; +using Crafter::ThreadPool; +using Crafter::Debug::Detail::Exchange; + +// Who may hold what, so nothing deadlocks: +// - A stream's callbacks (onData, onEnd, onSent) take only their exchange's +// own mutex, briefly, and never call into the session. +// - The session calls Stream under its own mutex, and Stream calls +// SendAsync under the exchange's, which Crafter.Network allows: SendAsync +// never reports from inside itself. +// - A stream wrapper is never destroyed, nor reset with Abort, under a +// mutex its callbacks take: destroying one waits for a callback running +// on another thread. +// - The handler, which takes the session's mutex, runs on a pool task, +// never on msquic's worker inside a callback. + +namespace { + // A request is a few hundred bytes. Past a megabyte the client has gone + // wrong, and the app would only buffer more of it. + constexpr std::size_t requestLimit = std::size_t{ 1 } << 20; + + // How far a `log --follow` client may fall behind, in bytes msquic has + // not yet taken. Past it the client is not reading, and the app would + // keep its backlog in memory for nothing, so its stream is reset. + constexpr std::uint64_t followLimit = std::uint64_t{ 8 } << 20; + + // The QUIC application error a lagging follower's stream is reset with. + constexpr std::uint64_t resetLagging = 1; + + // quit --force exits once its reply is sent, and "sent" means msquic + // took the bytes, not that they left: this gives them time to. + constexpr std::chrono::milliseconds exitGrace{ 200 }; + + QUICServerCredentials Credentials() { + QUICServerCredentials credentials; + // A per-user development certificate, which the client does not + // check: the listener is on loopback only, and what proves who asks + // is the token in each request, read from a file only this user can + // open. + credentials.selfSigned = true; + // A client killed without closing is let go after this, and its + // unanswered requests with it. + credentials.settings.idleTimeoutMs = 10'000; + return credentials; + } + + // One request's stream: reads the request, then answers on it. + class StreamExchange final : public Exchange, public std::enable_shared_from_this { + public: + using Handler = Detail::Server::Handler; + + explicit StreamExchange(QUICStream accepted) : stream(std::make_shared(std::move(accepted))) {} + + // Once, before anything else has the exchange. + void Receive(std::shared_ptr handler) { + const std::shared_ptr own = stream; + const std::shared_ptr self = shared_from_this(); + // The callbacks keep the exchange alive while the request + // arrives. QUICStream lets them go once the end is delivered, + // or once the wrapper is gone, which ends the cycle. + own->OnRecieve( + [self](std::span data) { self->Received(data); }, + [self, handler = std::move(handler)](bool clean) { self->Ended(clean, handler); }); + } + + void Reply(const Value& reply) override { + Finish(reply, nullptr); + } + + void ReplyThen(const Value& reply, std::function afterSent) override { + Finish(reply, std::move(afterSent)); + } + + bool Stream(const Value& line) override { + std::string text = line.Dump(); + text += '\n'; + std::shared_ptr lagging; + { + std::lock_guard lock(mutex); + if (!stream) return false; + if (stream->BytesInFlight() <= followLimit) { + // Under the lock, so lines leave in the order given. + const std::weak_ptr weak = weak_from_this(); + stream->SendAsync(text.data(), static_cast(text.size()), false, [weak](bool sent) { + if (sent) return; + if (const std::shared_ptr self = weak.lock()) self->Drop(); + }); + return true; + } + lagging = std::move(stream); + } + lagging->Abort(resetLagging); + return false; + } + + private: + std::mutex mutex; + // Until the reply is given or the client is gone. Shared, so a + // thread sending on it keeps it alive while another lets it go. + std::shared_ptr stream; + + // The callbacks' own, which never run two at once. + std::string request; + bool tooLarge = false; + + void Received(std::span data) { + if (tooLarge) return; + if (request.size() + data.size() > requestLimit) { + tooLarge = true; + request = std::string(); + Finish(MakeErrorReply(Codes::invalidRequest, std::format("a request is at most {} bytes", requestLimit)), nullptr); + return; + } + request.append(data.data(), data.size()); + } + + void Ended(bool clean, const std::shared_ptr& handler) { + if (tooLarge) return; + if (!clean) return Drop(); + // On a pool task, not here on msquic's worker: the handler takes + // the session's mutex, under which other threads send. + ThreadPool::Enqueue([self = shared_from_this(), handler, text = std::move(request)] { + self->Dispatch(text, *handler); + }); + } + + // On a pool task, where nothing may throw. + void Dispatch(const std::string& text, const Handler& handler) { + try { + Value parsed; + try { + parsed = Value::Parse(text); + } catch (const ParseError& error) { + return Reply(MakeErrorReply(Codes::invalidRequest, std::format("the request is not JSON: {}", error.what()))); + } + handler(parsed, shared_from_this()); + } catch (const std::exception& error) { + Reply(MakeErrorReply(Codes::exception, error.what())); + } catch (...) { + Reply(MakeErrorReply(Codes::exception, "the transport threw something that is not a std::exception")); + } + } + + // The first answer ends the exchange; any later one goes nowhere. + void Finish(const Value& reply, std::function afterSent) { + std::shared_ptr own; + { + std::lock_guard lock(mutex); + own = std::move(stream); + } + if (!own) { + // Nobody to tell. quit --force still exits. + if (afterSent) afterSent(); + return; + } + std::string text = reply.Dump(); + text += '\n'; + std::function onSent; + if (afterSent) { + onSent = [afterSent = std::move(afterSent)](bool) { + // Off msquic's worker, which still has the bytes to send. + try { + std::thread([afterSent] { + std::this_thread::sleep_for(exitGrace); + afterSent(); + }).detach(); + } catch (...) { + afterSent(); + } + }; + } + // With the FIN. Letting the wrapper go right after is fine: the + // send still goes out, and only the receive side is reset. + own->SendAsync(text.data(), static_cast(text.size()), true, std::move(onSent)); + } + + // The client is gone, or reset the request. + void Drop() { + std::shared_ptr own; + { + std::lock_guard lock(mutex); + own = std::move(stream); + } + } + }; +} + +struct Crafter::Debug::Detail::Server::State { + std::shared_ptr handler; + std::unique_ptr listener; + + // The accepted connections, each deleted by whoever takes it out under + // the mutex, and outside it: its OnClosed, or ~Server. + std::mutex mutex; + std::unordered_map> peers; + bool closing = false; +}; + +namespace { + using State = Crafter::Debug::Detail::Server::State; + + // A stream a client opened: one request. On the pool, where Crafter.Network + // delivers new streams. + void Serve(const std::weak_ptr& weak, QUICStream stream) { + try { + const std::shared_ptr state = weak.lock(); + if (!state) return; + std::make_shared(std::move(stream))->Receive(state->handler); + } catch (...) { + // Dropped: its client sees the stream reset. + } + } + + // The connection closed: the peer went, or its idle timeout passed. On + // msquic's worker, from inside the connection's own OnClosed, where + // deleting it is allowed. + void Forget(const std::weak_ptr& weak, ClientQUIC* closed) { + std::unique_ptr peer; + const std::shared_ptr state = weak.lock(); + if (!state) return; + { + std::lock_guard lock(state->mutex); + const auto found = state->peers.find(closed); + if (found == state->peers.end()) return; + peer = std::move(found->second); + state->peers.erase(found); + } + peer.reset(); + } + + // A new connection, on ListenerQUIC's accept thread, which nothing may + // leave by throwing. + void Accept(const std::weak_ptr& weak, ClientQUIC* accepted) { + std::unique_ptr peer(accepted); + try { + const std::shared_ptr state = weak.lock(); + if (!state) return; + peer->OnStream([weak](QUICStream stream) { Serve(weak, std::move(stream)); }); + { + std::lock_guard lock(state->mutex); + if (state->closing) return; + state->peers.emplace(accepted, std::move(peer)); + } + // Last, since from here its OnClosed may delete it at any time. + accepted->OnClosed([weak, accepted] { Forget(weak, accepted); }); + } catch (...) { + // Not served. A peer still owned here is deleted on the way out. + } + } +} + +Crafter::Debug::Detail::Server::Server(std::uint16_t requested, Handler handler) : state(std::make_shared()) { + state->handler = std::make_shared(std::move(handler)); + const std::weak_ptr weak = state; + state->listener = std::make_unique("127.0.0.1", requested, std::string(Crafter::Debug::alpn), Credentials(), + [weak](ClientQUIC* accepted) { Accept(weak, accepted); }); + port = state->listener->Port(); + state->listener->ListenAsyncSync(); +} + +Crafter::Debug::Detail::Server::~Server() { + { + std::lock_guard lock(state->mutex); + state->closing = true; + } + // Stops accepting and joins the accept loop: no Accept runs after this. + state->listener.reset(); + std::unordered_map> peers; + { + std::lock_guard lock(state->mutex); + peers.swap(state->peers); + } + // Outside the lock, which their OnClosed takes. Each waits, bounded, for + // its streams to finish, the replies just given among them, and for its + // close to be acknowledged. + peers.clear(); +} + +std::uint16_t Crafter::Debug::Detail::Server::Port() const { + return port; +} diff --git a/interfaces/Crafter.Debug-Transport.cppm b/interfaces/Crafter.Debug-Transport.cppm new file mode 100644 index 0000000..6bb48f1 --- /dev/null +++ b/interfaces/Crafter.Debug-Transport.cppm @@ -0,0 +1,49 @@ +// SPDX-License-Identifier: LGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// The app's end of the wire: QUIC on 127.0.0.1 with the ALPN +// crafter-debug/1, one request per stream the client opens. +// +// client {"token", "method", "params", "timeoutMs"}, then FIN +// server zero or more stream lines ({"entry": ...}, `log --follow`), +// then the reply ({"result": ...} or {"error": ...}), then FIN +// +// Every object is one line of JSON. Nothing here is exported: the session +// is what an app sees, and this partition keeps Crafter.Network's types out +// of the interface an app imports. +export module Crafter.Debug:Transport; +import std; +import Crafter.Debug.Core; +import :Session; + +namespace Crafter::Debug::Detail { + // Listens, and hands each complete request to the handler on a pool + // task, with an Exchange that answers on the request's own stream. + // + // Holds no thread while nothing arrives: Crafter.Network delivers bytes + // on msquic's workers and new streams on the host's pool, and replies go + // out without waiting. The one thread it has is ListenerQUIC's accept + // loop, which only registers each new connection's callbacks. + class Server { + public: + using Handler = std::function exchange)>; + + // On 127.0.0.1:port, 0 for any free port. Throws when it can't + // listen. Needs Crafter.Thread's pool running, with a worker. + Server(std::uint16_t port, Handler handler); + // Stops listening and closes every connection. Replies already + // given still reach their clients: each connection waits, bounded, + // for its streams to finish before it closes. + ~Server(); + Server(const Server&) = delete; + Server& operator=(const Server&) = delete; + + std::uint16_t Port() const; + + struct State; + + private: + std::shared_ptr state; + std::uint16_t port = 0; + }; +} diff --git a/interfaces/Crafter.Debug.cppm b/interfaces/Crafter.Debug.cppm index 290fa5c..fa51485 100644 --- a/interfaces/Crafter.Debug.cppm +++ b/interfaces/Crafter.Debug.cppm @@ -12,6 +12,9 @@ export import :Registry; export import :Timeline; export import :Condition; export import :Session; +// Exports nothing: the session's QUIC end, here only because every +// interface partition is the primary interface's to export. +export import :Transport; import std; export namespace Crafter::Debug { diff --git a/project.cpp b/project.cpp index c3af2fc..422d9d8 100644 --- a/project.cpp +++ b/project.cpp @@ -12,6 +12,10 @@ // debug build compiles it in the same pass as the app; an executable adds no // link flags to what depends on it. Consumers load this file through // Debug::Use (tooling/), which also writes their launcher. +// +// The library and the client talk QUIC through Crafter.Network, which comes +// from forgejo, with --local from the checkout beside this one, or from the +// project.cpp $CRAFTER_DEBUG_NETWORK names (a Network worktree). import std; import Crafter.Build; namespace fs = std::filesystem; @@ -30,6 +34,23 @@ namespace { return fs::path(std::source_location::current().file_name()).parent_path(); } + // Crafter.Network, with the args unchanged: a consumer that also uses + // it passes the same ones, so both resolve to one configuration rather + // than one module built twice. + Configuration* Network(std::span args) { + const std::vector forwarded(args.begin(), args.end()); + if (const char* local = std::getenv("CRAFTER_DEBUG_NETWORK"); local != nullptr && *local != '\0') { + return LocalProject({ .projectFile = local, .args = forwarded }); + } + if (std::ranges::find(args, std::string_view("--local")) != args.end()) { + return LocalProject({ .projectFile = Root().parent_path() / "Crafter.Network" / "project.cpp", .args = forwarded }); + } + return GitProject({ + .source = { .url = "https://forgejo.catcrafts.net/Catcrafts/Crafter.Network.git" }, + .args = forwarded, + }); + } + // Tests of the core, linked against it alone. constexpr std::array coreTests = { "ShouldRoundTripJson", @@ -118,6 +139,8 @@ extern "C" Configuration CrafterBuildProject(std::span a cfg.type = ConfigurationType::LibraryStatic; ApplyStandardArgs(cfg, args); + Configuration* network = Network(args); + Configuration& core = owned.emplace_back(); core.path = Root(); core.name = "Crafter.Debug.Core"; @@ -183,20 +206,22 @@ extern "C" Configuration CrafterBuildProject(std::span a client.GetInterfacesAndImplementations(interfaces, implementations); } - cfg.dependencies = { &core, &client }; + cfg.dependencies = { &core, &client, network }; { - std::array interfaces = { + std::array interfaces = { "interfaces/Crafter.Debug-Registry", "interfaces/Crafter.Debug-Timeline", "interfaces/Crafter.Debug-Condition", "interfaces/Crafter.Debug-Session", + "interfaces/Crafter.Debug-Transport", "interfaces/Crafter.Debug", }; - std::array implementations = { + std::array implementations = { "implementations/Crafter.Debug-Registry", "implementations/Crafter.Debug-Timeline", "implementations/Crafter.Debug-Condition", "implementations/Crafter.Debug-Session", + "implementations/Crafter.Debug-Transport", }; cfg.GetInterfacesAndImplementations(interfaces, implementations); }