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