Crafter.Debug/implementations/Crafter.Debug-Transport.cpp

// 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<StreamExchange> {
    public:
        using Handler = Detail::Server::Handler;

        explicit StreamExchange(QUICStream accepted) : stream(std::make_shared<QUICStream>(std::move(accepted))) {}

        // Once, before anything else has the exchange.
        void Receive(std::shared_ptr<const Handler> handler) {
            const std::shared_ptr<QUICStream> own = stream;
            const std::shared_ptr<StreamExchange> 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<const char> 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<void()> afterSent) override {
            Finish(reply, std::move(afterSent));
        }

        bool Stream(const Value& line) override {
            std::string text = line.Dump();
            text += '\n';
            std::shared_ptr<QUICStream> 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<StreamExchange> weak = weak_from_this();
                    stream->SendAsync(text.data(), static_cast<std::uint32_t>(text.size()), false, [weak](bool sent) {
                        if (sent) return;
                        if (const std::shared_ptr<StreamExchange> 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<QUICStream> stream;

        // The callbacks' own, which never run two at once.
        std::string request;
        bool tooLarge = false;

        void Received(std::span<const char> 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<const Handler>& 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<void()> afterSent) {
            std::shared_ptr<QUICStream> 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<void(bool)> 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<std::uint32_t>(text.size()), true, std::move(onSent));
        }

        // The client is gone, or reset the request.
        void Drop() {
            std::shared_ptr<QUICStream> own;
            {
                std::lock_guard lock(mutex);
                own = std::move(stream);
            }
        }
    };
}

struct Crafter::Debug::Detail::Server::State {
    std::shared_ptr<const Handler> handler;
    std::unique_ptr<ListenerQUIC> 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<ClientQUIC*, std::unique_ptr<ClientQUIC>> 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<State>& weak, QUICStream stream) {
        try {
            const std::shared_ptr<State> state = weak.lock();
            if (!state) return;
            std::make_shared<StreamExchange>(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<State>& weak, ClientQUIC* closed) {
        std::unique_ptr<ClientQUIC> peer;
        const std::shared_ptr<State> 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<State>& weak, ClientQUIC* accepted) {
        std::unique_ptr<ClientQUIC> peer(accepted);
        try {
            const std::shared_ptr<State> 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>()) {
    state->handler = std::make_shared<const Handler>(std::move(handler));
    const std::weak_ptr<State> weak = state;
    state->listener = std::make_unique<ListenerQUIC>("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<ClientQUIC*, std::unique_ptr<ClientQUIC>> 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;
}