From e60d6669d8959a9b6769ee35fb0e38ccfb655eeb Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 14:56:57 +0400 Subject: [PATCH] =?UTF-8?q?daemon:=20rpc/=20=E2=80=94=20Unix-socket=20tran?= =?UTF-8?q?sport=20+=20generated=20dispatch=20wiring=20(build=20step=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First real code in daemon/. veloxd now listens on $XDG_RUNTIME_DIR/velox/velox.sock (0600, SO_PEERCRED same-UID check), frames NDJSON, and routes every method through the generated velox::proto::dispatch(). The CLI and GUI have a server to talk to. Modules: - rpc/ndjson.hpp — newline-delimited framing, 8 MiB frame cap, CRLF- tolerant, partial-tail buffering. Header-only, tested. - rpc/event_loop — single-threaded poll(2) reactor; never blocks the loop. stop()/wake() are async-signal-safe (eventfd). - rpc/runtime_dir — $XDG_RUNTIME_DIR/velox resolution, 0700, owner-checked; refuses an insecure fallback rather than using /tmp. - rpc/uds_server — listener + non-blocking per-conn read/write with backpressure; handles session.hello (protocol-major check -> -32001, sessionId, transport=uds) and session.subscribe in the server layer; routes the rest through dispatch(). - rpc/dispatcher — VeloxDispatcher : proto::Dispatcher, all 39 methods. download.list answers an empty table; the rest return "not implemented" (-> -32603) until the store lands. - main.cpp — abstract-namespace single-instance lock, signal -> clean shutdown, socket unlinked on exit. Tests (ASan+UBSan and TSan clean): - veloxd.ndjson — framing edge cases - veloxd.uds_roundtrip — real socket: hello ok / version mismatch / empty list / -32601 / -32700 / pipelined requests, and a guard on the -32603 collapse documented in P1. Known gap, filed not worked around: daemon/docs/proto-requests-m1.md P1 — the generated Dispatcher has no error channel below -32603, so handlers cannot yet return -32010/-32011/-32013 with their data payloads. The server layer handles -32001/-32002/-32003 around dispatch(); genuine in-handler errors collapse to -32603 until PROTO gives handlers a real error return. Three error fixtures are non-conformant until then. Not in this drop: rpc/ws_server (next; needs the store for hashed pairing tokens), store/, sched/, cli/. WS reuses this event loop. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig --- daemon/CMakeLists.txt | 41 ++++ daemon/docs/proto-requests-m1.md | 79 +++++++ daemon/src/main.cpp | 98 +++++++++ daemon/src/rpc/.gitkeep | 0 daemon/src/rpc/dispatcher.cpp | 188 +++++++++++++++++ daemon/src/rpc/dispatcher.hpp | 103 ++++++++++ daemon/src/rpc/event_loop.cpp | 109 ++++++++++ daemon/src/rpc/event_loop.hpp | 69 +++++++ daemon/src/rpc/ndjson.hpp | 66 ++++++ daemon/src/rpc/runtime_dir.cpp | 56 +++++ daemon/src/rpc/runtime_dir.hpp | 30 +++ daemon/src/rpc/uds_server.cpp | 309 ++++++++++++++++++++++++++++ daemon/src/rpc/uds_server.hpp | 77 +++++++ daemon/src/version.hpp | 12 ++ daemon/tests/CMakeLists.txt | 14 ++ daemon/tests/check.hpp | 53 +++++ daemon/tests/ndjson_test.cpp | 64 ++++++ daemon/tests/uds_roundtrip_test.cpp | 195 ++++++++++++++++++ 18 files changed, 1563 insertions(+) create mode 100644 daemon/CMakeLists.txt create mode 100644 daemon/docs/proto-requests-m1.md create mode 100644 daemon/src/main.cpp delete mode 100644 daemon/src/rpc/.gitkeep create mode 100644 daemon/src/rpc/dispatcher.cpp create mode 100644 daemon/src/rpc/dispatcher.hpp create mode 100644 daemon/src/rpc/event_loop.cpp create mode 100644 daemon/src/rpc/event_loop.hpp create mode 100644 daemon/src/rpc/ndjson.hpp create mode 100644 daemon/src/rpc/runtime_dir.cpp create mode 100644 daemon/src/rpc/runtime_dir.hpp create mode 100644 daemon/src/rpc/uds_server.cpp create mode 100644 daemon/src/rpc/uds_server.hpp create mode 100644 daemon/src/version.hpp create mode 100644 daemon/tests/CMakeLists.txt create mode 100644 daemon/tests/check.hpp create mode 100644 daemon/tests/ndjson_test.cpp create mode 100644 daemon/tests/uds_roundtrip_test.cpp diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt new file mode 100644 index 0000000..6191528 --- /dev/null +++ b/daemon/CMakeLists.txt @@ -0,0 +1,41 @@ +# daemon/ produces the veloxd binary and the veloxd_rpc static library it is built from. +# Owned by lane DAEMON. Wired in by PKG via add_subdirectory(daemon) in the root file, +# guarded on this file existing. +# +# Layering (CLAUDE.md §3): depends on velox::core and velox::proto. No Qt. The engine +# (velox::core) is not linked yet — it arrives when sched/ and the task glue land; this +# first drop is the RPC transport + dispatcher skeleton so the CLI and GUI have a real +# server to talk to. + +if(NOT TARGET nlohmann_json::nlohmann_json) + find_package(nlohmann_json 3.11 REQUIRED) +endif() +find_package(Threads REQUIRED) + +# --- veloxd_rpc — the server library ---------------------------------------------------- +add_library(veloxd_rpc STATIC + src/rpc/runtime_dir.cpp + src/rpc/event_loop.cpp + src/rpc/uds_server.cpp + src/rpc/dispatcher.cpp +) +add_library(velox::daemon_rpc ALIAS veloxd_rpc) + +target_include_directories(veloxd_rpc + PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src +) +target_compile_features(veloxd_rpc PUBLIC cxx_std_23) +target_compile_options(veloxd_rpc PRIVATE -Wall -Wextra -Wpedantic -Werror) +target_link_libraries(veloxd_rpc + PUBLIC velox::proto nlohmann_json::nlohmann_json Threads::Threads +) + +# --- veloxd — the daemon binary ------------------------------------------------------- +add_executable(veloxd src/main.cpp) +target_compile_features(veloxd PRIVATE cxx_std_23) +target_compile_options(veloxd PRIVATE -Wall -Wextra -Wpedantic -Werror) +target_link_libraries(veloxd PRIVATE veloxd_rpc) + +if(VELOX_BUILD_TESTS AND EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt) + add_subdirectory(tests) +endif() diff --git a/daemon/docs/proto-requests-m1.md b/daemon/docs/proto-requests-m1.md new file mode 100644 index 0000000..0833155 --- /dev/null +++ b/daemon/docs/proto-requests-m1.md @@ -0,0 +1,79 @@ +# DAEMON → PROTO — requests against `contracts/` (and its codegen) + +Status: **open**. Raised by lane DAEMON while building `rpc/` against `1.3.0`. +PROTO owns `contracts/`, including `contracts/codegen/`. Ranking per +`contracts/README.md` rule 4: a codegen output-shape change that every server must +adopt is effectively **major for the C++ binding** even when the wire is untouched — +it needs a version note and a regen, not a silent change. + +--- + +## P1. The generated `Dispatcher` has no error channel below `-32603` — **blocking a conformant server** + +`velox::proto::Dispatcher`'s 39 methods each return `Result` = +`std::expected`, and `dispatch()` maps **every** handler error to +`ErrorCode::InternalError` (`-32603`): + +```cpp +auto r = handler.on_download_get(*p); +if (!r) + return make_error(id, ErrorCode::InternalError, r.error().message, + nlohmann::json{{"path", r.error().path}}); +``` + +So a handler cannot return any of the contract's own error codes. The error fixtures +in `contracts/fixtures/errors/` that a live server must satisfy (DAEMON DoD: "passes +the full conformance suite as a server, over both transports") include: + +| Fixture | Expected code | `data` | Originates | +|---|---|---|---| +| `download.get.not-found` | `-32010` | `{taskId}` | inside the handler | +| `download.add.invalid-path` | `-32011` | `{path}` | inside the handler (after canonicalization) | +| `download.probe.probe-failed` | `-32013` | `{httpStatus}` | inside the handler | +| `session.pair.rate-limited` | `-32014` | `{retryAfterSec}` | server-side gate, but cleanest expressed as a handler result | +| `session.hello.version-mismatch` | `-32001` | `{expected, actual}` | can be done server-side around `dispatch()` | +| `session.hello.not-paired` | `-32002` | — | server-side WS auth gate, around `dispatch()` | + +`-32001`, `-32002`, `-32003` DAEMON can and will handle in the server layer that wraps +`dispatch()` (`-32003` is already in `dispatch()` itself). But `-32010`, `-32011`, +`-32013` are per-method **handler outcomes** — the daemon knows "no such task" only +after the store lookup, "outside allowed roots" only after `realpath()`. There is no +correct way to surface them today except misreporting as `-32603`, which the TS +conformance replay will reject on the `code` compare. + +**Requested:** give the generated handler methods an error return that carries an +`ErrorCode`, a message, and a free-form `data` object. Shape is PROTO's call; a +minimal one that keeps `ParseError` for the parse path and adds a handler-error type: + +```cpp +struct HandlerError { + ErrorCode code{ErrorCode::InternalError}; + std::string message; + nlohmann::json data{nullptr}; +}; +template using HandlerResult = std::expected; +// Dispatcher::on_* return HandlerResult; dispatch() forwards code/message/data +// straight into make_error() instead of hard-coding InternalError. +``` + +`FixtureDispatcher` and `conformance_main.cpp` would need the trivial follow-on edit +(they only ever return success today, so it is a type-name swap). + +Until this lands, DAEMON's `rpc/` server layer handles `-3200x` around `dispatch()` +where it can, and every genuine in-handler failure collapses to `-32603` with a clear +message — visibly non-conformant on three error fixtures, tracked here, not worked +around by inventing a side channel. + +--- + +## P2. `SessionHelloResult.transport` and `-32001` `data.expected` — minor clarifications + +- `session.hello.version-mismatch`'s `data.expected` is `"1.0.0"` in the fixture, i.e. + the daemon's *current* protocol version string, not a bare major. DAEMON will echo + `kProtocolVersion` (`"1.3.0"`) there unless PROTO wants the fixture's literal + `"1.0.0"` preserved — flag if the conformance compare is exact on that field rather + than structural. +- `SessionHelloResult.transport` is `std::optional` — DAEMON intends to always populate + it (`"uds"` / `"ws"`) so a client knows its privilege level up front, as the field's + own description invites. No change requested; noting the intent so a later "why is + this always set" review has the answer. diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp new file mode 100644 index 0000000..e410538 --- /dev/null +++ b/daemon/src/main.cpp @@ -0,0 +1,98 @@ +// veloxd — the Velox download-manager daemon. +// +// This drop wires up the Unix-socket RPC transport and a dispatcher skeleton so the CLI +// and GUI have a real server to speak to (AGENT-DAEMON.md build order, step 1). The +// WebSocket transport, the SQLite store and the scheduler land next. + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "rpc/dispatcher.hpp" +#include "rpc/event_loop.hpp" +#include "rpc/runtime_dir.hpp" +#include "rpc/uds_server.hpp" +#include "version.hpp" + +namespace { + +velox::daemon::rpc::EventLoop* g_loop = nullptr; + +void on_signal(int) { + if (g_loop != nullptr) g_loop->stop(); // stop() is async-signal-safe (writes an eventfd) +} + +// Single-instance guard: bind an abstract-namespace Unix socket whose name is unique to +// this user. A second daemon gets EADDRINUSE and exits. The kernel reclaims an +// abstract-namespace address when the holding process dies, so a crash never wedges it +// (docs/01 §2). Returns the held fd (kept open for the process lifetime) or -1. +int acquire_single_instance_lock() { + const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (fd < 0) return -1; + + const std::string name = std::string("velox-daemon-") + std::to_string(::geteuid()); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + // Leading NUL selects the abstract namespace; the name follows, not NUL-terminated. + addr.sun_path[0] = '\0'; + std::memcpy(addr.sun_path + 1, name.c_str(), name.size()); + const socklen_t len = + static_cast(offsetof(sockaddr_un, sun_path) + 1 + name.size()); + + if (::bind(fd, reinterpret_cast(&addr), len) != 0) { + ::close(fd); + return -1; + } + return fd; +} + +} // namespace + +int main() { + std::cout << "veloxd " << velox::daemon::kDaemonVersion << " (protocol " + << velox::proto::kProtocolVersion << ")\n"; + + const int lock_fd = acquire_single_instance_lock(); + if (lock_fd < 0) { + std::cerr << "veloxd: another instance is already running for this user\n"; + return 1; + } + + velox::daemon::rpc::RuntimeDir rt; + if (const auto ec = velox::daemon::rpc::resolve_runtime_dir(rt)) { + std::cerr << "veloxd: cannot prepare runtime directory: " << ec.message() << "\n"; + return 1; + } + + velox::daemon::rpc::EventLoop loop; + g_loop = &loop; + + struct sigaction sa{}; + sa.sa_handler = on_signal; + ::sigemptyset(&sa.sa_mask); + ::sigaction(SIGINT, &sa, nullptr); + ::sigaction(SIGTERM, &sa, nullptr); + ::signal(SIGPIPE, SIG_IGN); // a client vanishing mid-write is EPIPE, never a signal + + velox::daemon::rpc::VeloxDispatcher dispatcher; + velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path()); + if (const auto ec = uds.start()) { + std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message() + << "\n"; + return 1; + } + + std::cout << "veloxd: listening on " << uds.socket_path() << "\n"; + loop.run(); + std::cout << "veloxd: shutting down\n"; + + g_loop = nullptr; + ::close(lock_fd); + return 0; +} diff --git a/daemon/src/rpc/.gitkeep b/daemon/src/rpc/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp new file mode 100644 index 0000000..4063387 --- /dev/null +++ b/daemon/src/rpc/dispatcher.cpp @@ -0,0 +1,188 @@ +#include "rpc/dispatcher.hpp" + +namespace velox::daemon::rpc { + +namespace proto = velox::proto; + +namespace { + +// The generated Result carries only proto::ParseError, whose {path, message} the +// generated dispatch() forwards as -32603 data. Until P1 (daemon/docs/proto-requests-m1.md) +// gives handlers a real error channel, an unimplemented method says so plainly here. +template +proto::Result not_implemented(const char* method) { + return std::unexpected(proto::ParseError{method, "not implemented in this build"}); +} + +} // namespace + +// --- session.* : handled in the server layer, unreachable here in the running daemon --- +// Kept as explicit stubs so a direct dispatch() caller (a test, a future in-process client) +// gets a clear answer rather than undefined behaviour from a missing override. + +proto::Result +VeloxDispatcher::on_session_hello(const proto::SessionHelloParams&) { + return not_implemented("session.hello"); +} + +proto::Result +VeloxDispatcher::on_session_pair(const proto::SessionPairParams&) { + return not_implemented("session.pair"); +} + +proto::Result +VeloxDispatcher::on_session_subscribe(const proto::SessionSubscribeParams&) { + return not_implemented("session.subscribe"); +} + +// --- download.list : an empty table, so a client can connect and render --------------- + +proto::Result +VeloxDispatcher::on_download_list(const proto::DownloadListParams&) { + proto::DownloadListResult r; + r.total = 0; + return r; +} + +// --- everything else : not implemented until the store and scheduler land ------------- + +proto::Result +VeloxDispatcher::on_capture_getRules(const proto::CaptureGetRulesParams&) { + return not_implemented("capture.getRules"); +} +proto::Result +VeloxDispatcher::on_capture_offer(const proto::CaptureOfferParams&) { + return not_implemented("capture.offer"); +} +proto::Result +VeloxDispatcher::on_category_list(const proto::CategoryListParams&) { + return not_implemented("category.list"); +} +proto::Result +VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) { + return not_implemented("category.remove"); +} +proto::Result +VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) { + return not_implemented("category.upsert"); +} +proto::Result +VeloxDispatcher::on_download_add(const proto::DownloadSpec&) { + return not_implemented("download.add"); +} +proto::Result +VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) { + return not_implemented("download.addBatch"); +} +proto::Result +VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) { + return not_implemented("download.cancel"); +} +proto::Result +VeloxDispatcher::on_download_get(const proto::DownloadGetParams&) { + return not_implemented("download.get"); +} +proto::Result +VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) { + return not_implemented("download.pause"); +} +proto::Result +VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) { + return not_implemented("download.probe"); +} +proto::Result +VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams&) { + return not_implemented("download.provideAuth"); +} +proto::Result +VeloxDispatcher::on_download_refreshUrl(const proto::DownloadRefreshUrlParams&) { + return not_implemented("download.refreshUrl"); +} +proto::Result +VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) { + return not_implemented("download.remove"); +} +proto::Result +VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams&) { + return not_implemented("download.resume"); +} +proto::Result +VeloxDispatcher::on_download_start(const proto::DownloadStartParams&) { + return not_implemented("download.start"); +} +proto::Result +VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams&) { + return not_implemented("download.update"); +} +proto::Result +VeloxDispatcher::on_grabber_harvest(const proto::GrabberHarvestParams&) { + return not_implemented("grabber.harvest"); +} +proto::Result +VeloxDispatcher::on_grabber_start(const proto::GrabberStartParams&) { + return not_implemented("grabber.start"); +} +proto::Result +VeloxDispatcher::on_grabber_status(const proto::GrabberStatusParams&) { + return not_implemented("grabber.status"); +} +proto::Result VeloxDispatcher::on_limiter_get(const proto::LimiterGetParams&) { + return not_implemented("limiter.get"); +} +proto::Result VeloxDispatcher::on_limiter_set(const proto::Limiter&) { + return not_implemented("limiter.set"); +} +proto::Result +VeloxDispatcher::on_media_addVariant(const proto::MediaAddVariantParams&) { + return not_implemented("media.addVariant"); +} +proto::Result +VeloxDispatcher::on_media_listVariants(const proto::MediaListVariantsParams&) { + return not_implemented("media.listVariants"); +} +proto::Result +VeloxDispatcher::on_queue_list(const proto::QueueListParams&) { + return not_implemented("queue.list"); +} +proto::Result +VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) { + return not_implemented("queue.reorder"); +} +proto::Result +VeloxDispatcher::on_queue_start(const proto::QueueStartParams&) { + return not_implemented("queue.start"); +} +proto::Result +VeloxDispatcher::on_queue_stop(const proto::QueueStopParams&) { + return not_implemented("queue.stop"); +} +proto::Result +VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) { + return not_implemented("queue.upsert"); +} +proto::Result +VeloxDispatcher::on_rules_list(const proto::RulesListParams&) { + return not_implemented("rules.list"); +} +proto::Result +VeloxDispatcher::on_rules_upsert(const proto::RulesUpsertParams&) { + return not_implemented("rules.upsert"); +} +proto::Result +VeloxDispatcher::on_schedule_get(const proto::ScheduleGetParams&) { + return not_implemented("schedule.get"); +} +proto::Result +VeloxDispatcher::on_schedule_set(const proto::ScheduleSetParams&) { + return not_implemented("schedule.set"); +} +proto::Result +VeloxDispatcher::on_settings_get(const proto::SettingsGetParams&) { + return not_implemented("settings.get"); +} +proto::Result +VeloxDispatcher::on_settings_set(const proto::SettingsSetParams&) { + return not_implemented("settings.set"); +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp new file mode 100644 index 0000000..5e7e97a --- /dev/null +++ b/daemon/src/rpc/dispatcher.hpp @@ -0,0 +1,103 @@ +#pragma once + +// VeloxDispatcher implements the generated velox::proto::Dispatcher — one virtual per RPC +// method. The generated dispatch() does the envelope, the transport check and the param +// parse; a method here only ever sees a validated, typed params struct and returns a +// typed result. +// +// Scope of this drop (AGENT-DAEMON.md build order): the transport is real, the store is +// not. session.hello / session.pair / session.subscribe are handled in the server layer +// (they are connection- and transport-stateful) and never reach this class in the running +// daemon. download.list answers with an empty table so a client can connect and render. +// Every other method returns "not implemented in this build" — which the generated +// dispatch() surfaces as -32603 — until the store and scheduler land. +// +// The -32603 collapse for genuine in-handler errors (-32010 / -32011 / -32013) is a known +// codegen gap, filed as P1 in daemon/docs/proto-requests-m1.md. Not worked around here. + +#include "velox_proto.hpp" + +namespace velox::daemon::rpc { + +class VeloxDispatcher final : public velox::proto::Dispatcher { +public: + velox::proto::Result + on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override; + velox::proto::Result + on_capture_offer(const velox::proto::CaptureOfferParams&) override; + velox::proto::Result + on_category_list(const velox::proto::CategoryListParams&) override; + velox::proto::Result + on_category_remove(const velox::proto::CategoryRemoveParams&) override; + velox::proto::Result + on_category_upsert(const velox::proto::CategoryUpsertParams&) override; + velox::proto::Result + on_download_add(const velox::proto::DownloadSpec&) override; + velox::proto::Result + on_download_addBatch(const velox::proto::DownloadAddBatchParams&) override; + velox::proto::Result + on_download_cancel(const velox::proto::DownloadCancelParams&) override; + velox::proto::Result + on_download_get(const velox::proto::DownloadGetParams&) override; + velox::proto::Result + on_download_list(const velox::proto::DownloadListParams&) override; + velox::proto::Result + on_download_pause(const velox::proto::DownloadPauseParams&) override; + velox::proto::Result + on_download_probe(const velox::proto::DownloadProbeParams&) override; + velox::proto::Result + on_download_provideAuth(const velox::proto::DownloadProvideAuthParams&) override; + velox::proto::Result + on_download_refreshUrl(const velox::proto::DownloadRefreshUrlParams&) override; + velox::proto::Result + on_download_remove(const velox::proto::DownloadRemoveParams&) override; + velox::proto::Result + on_download_resume(const velox::proto::DownloadResumeParams&) override; + velox::proto::Result + on_download_start(const velox::proto::DownloadStartParams&) override; + velox::proto::Result + on_download_update(const velox::proto::DownloadUpdateParams&) override; + velox::proto::Result + on_grabber_harvest(const velox::proto::GrabberHarvestParams&) override; + velox::proto::Result + on_grabber_start(const velox::proto::GrabberStartParams&) override; + velox::proto::Result + on_grabber_status(const velox::proto::GrabberStatusParams&) override; + velox::proto::Result + on_limiter_get(const velox::proto::LimiterGetParams&) override; + velox::proto::Result on_limiter_set(const velox::proto::Limiter&) override; + velox::proto::Result + on_media_addVariant(const velox::proto::MediaAddVariantParams&) override; + velox::proto::Result + on_media_listVariants(const velox::proto::MediaListVariantsParams&) override; + velox::proto::Result + on_queue_list(const velox::proto::QueueListParams&) override; + velox::proto::Result + on_queue_reorder(const velox::proto::QueueReorderParams&) override; + velox::proto::Result + on_queue_start(const velox::proto::QueueStartParams&) override; + velox::proto::Result + on_queue_stop(const velox::proto::QueueStopParams&) override; + velox::proto::Result + on_queue_upsert(const velox::proto::QueueUpsertParams&) override; + velox::proto::Result + on_rules_list(const velox::proto::RulesListParams&) override; + velox::proto::Result + on_rules_upsert(const velox::proto::RulesUpsertParams&) override; + velox::proto::Result + on_schedule_get(const velox::proto::ScheduleGetParams&) override; + velox::proto::Result + on_schedule_set(const velox::proto::ScheduleSetParams&) override; + velox::proto::Result + on_session_hello(const velox::proto::SessionHelloParams&) override; + velox::proto::Result + on_session_pair(const velox::proto::SessionPairParams&) override; + velox::proto::Result + on_session_subscribe(const velox::proto::SessionSubscribeParams&) override; + velox::proto::Result + on_settings_get(const velox::proto::SettingsGetParams&) override; + velox::proto::Result + on_settings_set(const velox::proto::SettingsSetParams&) override; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/event_loop.cpp b/daemon/src/rpc/event_loop.cpp new file mode 100644 index 0000000..3713b36 --- /dev/null +++ b/daemon/src/rpc/event_loop.cpp @@ -0,0 +1,109 @@ +#include "rpc/event_loop.hpp" + +#include +#include +#include + +#include +#include +#include +#include + +namespace velox::daemon::rpc { + +EventLoop::EventLoop() { + wake_fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + if (wake_fd_ < 0) throw std::runtime_error("eventfd() failed"); + fds_.emplace(wake_fd_, Entry{kRead, [this](int, unsigned) { drain_wakeup(); }}); +} + +EventLoop::~EventLoop() { + if (wake_fd_ >= 0) ::close(wake_fd_); +} + +void EventLoop::add_fd(int fd, unsigned interest, Callback cb) { + fds_[fd] = Entry{interest, std::move(cb)}; +} + +void EventLoop::mod_fd(int fd, unsigned interest) { + if (auto it = fds_.find(fd); it != fds_.end()) it->second.interest = interest; +} + +void EventLoop::del_fd(int fd) { + if (fd == wake_fd_) return; // internal, never removed + fds_.erase(fd); +} + +void EventLoop::wake() noexcept { + const std::uint64_t one = 1; + // Best-effort: an EAGAIN here means a wakeup is already pending, which is fine. + [[maybe_unused]] ssize_t n = ::write(wake_fd_, &one, sizeof(one)); +} + +void EventLoop::stop() noexcept { + stop_requested_ = true; + wake(); +} + +void EventLoop::drain_wakeup() noexcept { + std::uint64_t sink = 0; + while (::read(wake_fd_, &sink, sizeof(sink)) > 0) { + } +} + +void EventLoop::run() { + if (running_) throw std::logic_error("EventLoop::run() is not re-entrant"); + running_ = true; + stop_requested_ = false; + + std::vector pfds; + std::vector fired; + + while (!stop_requested_) { + pfds.clear(); + pfds.reserve(fds_.size()); + for (const auto& [fd, e] : fds_) { + short ev = 0; + if (e.interest & kRead) ev |= POLLIN; + if (e.interest & kWrite) ev |= POLLOUT; + if (ev == 0 && fd != wake_fd_) continue; + pollfd p{}; + p.fd = fd; + p.events = ev; + pfds.push_back(p); + } + + const int rc = ::poll(pfds.data(), pfds.size(), -1); + if (rc < 0) { + if (errno == EINTR) continue; + throw std::runtime_error("poll() failed"); + } + if (rc == 0) continue; + + // Snapshot the fds that fired before invoking any callback: a callback may erase + // entries from fds_, which would invalidate iteration over pfds' referents. + fired.clear(); + for (const auto& p : pfds) { + if (p.revents != 0) fired.push_back(p.fd); + } + + for (const int fd : fired) { + const auto it = fds_.find(fd); + if (it == fds_.end()) continue; // removed by an earlier callback this pass + + // Recompute revents for this fd from the snapshot. + unsigned events = 0; + for (const auto& p : pfds) { + if (p.fd != fd) continue; + if (p.revents & (POLLIN | POLLHUP | POLLERR)) events |= kRead; + if (p.revents & POLLOUT) events |= kWrite; + break; + } + if (events != 0) it->second.cb(fd, events); + } + } + + running_ = false; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/event_loop.hpp b/daemon/src/rpc/event_loop.hpp new file mode 100644 index 0000000..8f96b4a --- /dev/null +++ b/daemon/src/rpc/event_loop.hpp @@ -0,0 +1,69 @@ +#pragma once + +// A single-threaded poll(2) reactor. Every RPC listener and connection registers its fd +// here; the loop never blocks on disk or DNS (AGENT-DAEMON.md build step 1 — "Never block +// the RPC loop"). Long work is handed to CORE's pools later; this class only multiplexes +// readiness. +// +// Thread model: run() executes on one thread. add_fd/mod_fd/del_fd are called from +// callbacks on that same thread. stop() and wake() are async-signal-safe and safe to call +// from any thread or a signal handler — they only write() a byte to an internal eventfd. + +#include +#include +#include +#include + +namespace velox::daemon::rpc { + +enum Interest : unsigned { + kNone = 0, + kRead = 1u << 0, + kWrite = 1u << 1, +}; + +class EventLoop { +public: + // Called when the fd is readable and/or writable. `events` is the subset of the fd's + // registered Interest that fired. A callback may add/modify/remove any fd, including + // its own, and may call stop(). + using Callback = std::function; + + EventLoop(); + ~EventLoop(); + + EventLoop(const EventLoop&) = delete; + EventLoop& operator=(const EventLoop&) = delete; + + // Register `fd` (must be non-blocking) for `interest`. Replaces any prior registration. + void add_fd(int fd, unsigned interest, Callback cb); + // Change the interest mask for an already-registered fd. + void mod_fd(int fd, unsigned interest); + // Stop watching `fd`. Does not close it — ownership stays with the caller. + void del_fd(int fd); + + // Run until stop() is called. Re-entrant calls are not supported. + void run(); + + // Ask run() to return after the current poll wakeup. Async-signal-safe. + void stop() noexcept; + + // Force one poll() wakeup without stopping — used when interest changed from outside a + // callback. Async-signal-safe. + void wake() noexcept; + +private: + struct Entry { + unsigned interest; + Callback cb; + }; + + void drain_wakeup() noexcept; + + int wake_fd_; // eventfd, always registered + bool running_ = false; + std::atomic stop_requested_ = false; // set from stop(), read by run() + std::unordered_map fds_; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ndjson.hpp b/daemon/src/rpc/ndjson.hpp new file mode 100644 index 0000000..ef79b23 --- /dev/null +++ b/daemon/src/rpc/ndjson.hpp @@ -0,0 +1,66 @@ +#pragma once + +// NDJSON framing: one JSON value per line, '\n'-terminated. This is the wire framing on +// the Unix socket ($XDG_RUNTIME_DIR/velox/velox.sock) per AGENT-DAEMON.md build step 1. +// A frame carries no length prefix — the newline is the delimiter — so a reader must +// buffer a partial tail until the next '\n' arrives. +// +// Header-only: it is pure string slicing with no I/O and no dependency beyond . + +#include +#include +#include +#include + +namespace velox::daemon::rpc { + +// Largest single frame accepted before the connection is considered abusive. A well-formed +// request (even download.addBatch with a big clipboard blob) is far below this; anything +// past it is either a bug or an attack, and the server drops the connection. +inline constexpr std::size_t kMaxFrameBytes = 8 * 1024 * 1024; + +// Accumulates bytes off a stream socket and hands back complete lines. Bytes after the +// last '\n' stay buffered for next time. A trailing '\r' (CRLF) is trimmed so a client +// that writes CRLF still parses. +class FrameReader { +public: + // Feed a chunk just read from the socket. Returns the frames completed by this chunk, + // in order, each with its line terminator removed. Empty lines are skipped (a stray + // blank line between frames is not an error). + std::vector feed(std::string_view chunk) { + std::vector out; + buf_.append(chunk); + std::size_t start = 0; + for (;;) { + const std::size_t nl = buf_.find('\n', start); + if (nl == std::string::npos) break; + std::string_view line{buf_.data() + start, nl - start}; + if (!line.empty() && line.back() == '\r') line.remove_suffix(1); + if (!line.empty()) out.emplace_back(line); + start = nl + 1; + } + buf_.erase(0, start); + return out; + } + + // True once the unframed tail has grown past the cap without a newline — the caller + // must close the connection rather than buffer without bound. + bool overflowed() const noexcept { return buf_.size() > kMaxFrameBytes; } + + std::size_t buffered() const noexcept { return buf_.size(); } + +private: + std::string buf_; +}; + +// Frame a payload for writing: exactly the JSON text plus one '\n'. Kept as a function so +// the "+ newline" rule lives in one place. +inline std::string frame(std::string_view payload) { + std::string out; + out.reserve(payload.size() + 1); + out.append(payload); + out.push_back('\n'); + return out; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/runtime_dir.cpp b/daemon/src/rpc/runtime_dir.cpp new file mode 100644 index 0000000..e387916 --- /dev/null +++ b/daemon/src/rpc/runtime_dir.cpp @@ -0,0 +1,56 @@ +#include "rpc/runtime_dir.hpp" + +#include +#include +#include + +#include +#include +#include + +namespace velox::daemon::rpc { + +namespace { + +std::error_code errc(int e) { return std::error_code(e, std::generic_category()); } + +// Ensure `dir` exists as a directory we own with mode 0700. Creates it if absent. +std::error_code ensure_private_dir(const std::string& dir) { + if (::mkdir(dir.c_str(), 0700) != 0 && errno != EEXIST) return errc(errno); + + struct stat st{}; + if (::lstat(dir.c_str(), &st) != 0) return errc(errno); + if (!S_ISDIR(st.st_mode)) return errc(ENOTDIR); + if (st.st_uid != ::geteuid()) return errc(EPERM); + + // Tighten if a prior run (or umask) left it looser. Group/other bits must be clear: + // the socket is 0600 but a traversable parent still lets another user stat it. + if ((st.st_mode & 077) != 0 && ::chmod(dir.c_str(), 0700) != 0) return errc(errno); + return {}; +} + +} // namespace + +std::error_code resolve_runtime_dir(RuntimeDir& out) { + std::string base; + if (const char* xdg = ::getenv("XDG_RUNTIME_DIR"); xdg != nullptr && xdg[0] != '\0') { + base = xdg; + } else { + base = "/run/user/" + std::to_string(::geteuid()); + struct stat st{}; + if (::stat(base.c_str(), &st) != 0 || !S_ISDIR(st.st_mode)) { + // No XDG_RUNTIME_DIR and no /run/user/: we refuse rather than pick an + // insecure fallback. The caller surfaces this as "cannot start". + return errc(ENOENT); + } + } + if (!base.empty() && base.back() == '/') base.pop_back(); + + const std::string dir = base + "/velox"; + if (auto ec = ensure_private_dir(dir)) return ec; + + out.path = dir; + return {}; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/runtime_dir.hpp b/daemon/src/rpc/runtime_dir.hpp new file mode 100644 index 0000000..b71a529 --- /dev/null +++ b/daemon/src/rpc/runtime_dir.hpp @@ -0,0 +1,30 @@ +#pragma once + +// Resolves $XDG_RUNTIME_DIR/velox/ — the home of velox.sock, ws.port and the +// single-instance lock (docs/01 §5, AGENT-DAEMON.md build step 1). Creating it 0700 and +// refusing a pre-existing dir we do not own is a security boundary, not a convenience. + +#include +#include + +namespace velox::daemon::rpc { + +struct RuntimeDir { + std::string path; // absolute, no trailing slash, e.g. /run/user/1000/velox + + std::string socket_path() const { return path + "/velox.sock"; } + std::string ws_port_path() const { return path + "/ws.port"; } +}; + +// Resolve and ensure the directory exists, mode 0700, owned by the current user. +// +// - $XDG_RUNTIME_DIR set -> "/velox" +// - unset -> "/run/user//velox" if that base exists, else an error +// (we do not fall back to /tmp: a world-traversable runtime dir defeats the 0600 socket) +// +// On success `out` is filled and an ok error_code is returned. On failure `out` is +// untouched and the error_code explains why (base missing, exists but not a dir, wrong +// owner, wrong perms, mkdir failed). +std::error_code resolve_runtime_dir(RuntimeDir& out); + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp new file mode 100644 index 0000000..9437090 --- /dev/null +++ b/daemon/src/rpc/uds_server.cpp @@ -0,0 +1,309 @@ +#include "rpc/uds_server.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include + +#include + +#include "rpc/event_loop.hpp" +#include "version.hpp" + +namespace velox::daemon::rpc { + +namespace proto = velox::proto; +using nlohmann::json; + +namespace { + +std::error_code errc(int e) { return std::error_code(e, std::generic_category()); } + +// Largest reply we will buffer for a client that is not reading. Past this the client is +// wedged and the connection is dropped rather than growing the daemon's RSS without bound. +constexpr std::size_t kMaxOutBytes = 16 * 1024 * 1024; + +std::string uuid4() { + std::random_device rd; + std::uniform_int_distribution d; + std::uint32_t a = d(rd), b = d(rd), c = d(rd), e = d(rd); + b = (b & 0xFFFF0FFFu) | 0x00004000u; // version 4 + c = (c & 0x3FFFFFFFu) | 0x80000000u; // variant 1 + char buf[37]; + std::snprintf(buf, sizeof(buf), "%08x-%04x-%04x-%04x-%04x%08x", a, (b >> 16), (b & 0xFFFF), + (c >> 16), (c & 0xFFFF), e); + return std::string(buf); +} + +int major_of(const std::string& semver) { + // "1.3.0" -> 1. A missing or non-numeric leading component is treated as major -1 so + // it can never accidentally match the daemon's. + try { + return std::stoi(semver.substr(0, semver.find('.'))); + } catch (...) { + return -1; + } +} + +json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json data = nullptr) { + return proto::make_error(id, code, msg, std::move(data)); +} + +} // namespace + +UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, std::string socket_path) + : loop_(loop), dispatcher_(dispatcher), path_(std::move(socket_path)) {} + +UdsServer::~UdsServer() { + for (auto& [fd, c] : conns_) { + loop_.del_fd(fd); + ::close(fd); + } + if (listen_fd_ >= 0) { + loop_.del_fd(listen_fd_); + ::close(listen_fd_); + } + if (bound_) ::unlink(path_.c_str()); +} + +std::error_code UdsServer::start() { + if (path_.size() + 1 > sizeof(sockaddr_un::sun_path)) return errc(ENAMETOOLONG); + + const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); + if (fd < 0) return errc(errno); + + // A socket file from a previous run blocks bind() with EADDRINUSE. Single-instance is + // enforced separately (main.cpp lock socket), so an existing file here is stale. + ::unlink(path_.c_str()); + + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + std::memcpy(addr.sun_path, path_.c_str(), path_.size()); + + // umask can only tighten; set the mode explicitly after bind so it is exactly 0600. + if (::bind(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + const int e = errno; + ::close(fd); + return errc(e); + } + bound_ = true; + if (::chmod(path_.c_str(), 0600) != 0) { + const int e = errno; + ::close(fd); + ::unlink(path_.c_str()); + bound_ = false; + return errc(e); + } + if (::listen(fd, SOMAXCONN) != 0) { + const int e = errno; + ::close(fd); + ::unlink(path_.c_str()); + bound_ = false; + return errc(e); + } + + listen_fd_ = fd; + loop_.add_fd(listen_fd_, kRead, [this](int, unsigned) { on_listener_readable(); }); + return {}; +} + +void UdsServer::on_listener_readable() { + for (;;) { + const int cfd = ::accept4(listen_fd_, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); + if (cfd < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + if (errno == EINTR || errno == ECONNABORTED) continue; + break; // EMFILE/ENFILE: stop accepting this pass; loop retries on next readable + } + + ucred cred{}; + socklen_t len = sizeof(cred); + if (::getsockopt(cfd, SOL_SOCKET, SO_PEERCRED, &cred, &len) != 0 || + cred.uid != ::geteuid()) { + // Not the same user. The socket mode should already prevent this; refuse hard + // regardless — this is the authorization on the Unix transport (docs/01 §2). + ::close(cfd); + continue; + } + + auto conn = std::make_unique(); + conn->fd = cfd; + conns_.emplace(cfd, std::move(conn)); + loop_.add_fd(cfd, kRead, [this](int fd, unsigned ev) { on_conn_event(fd, ev); }); + } +} + +void UdsServer::on_conn_event(int fd, unsigned events) { + const auto it = conns_.find(fd); + if (it == conns_.end()) return; + Conn& c = *it->second; + + if (events & kWrite) { + flush(c); + if (conns_.find(fd) == conns_.end()) return; // flush closed it + } + + if (events & kRead) { + char buf[64 * 1024]; + for (;;) { + const ssize_t n = ::read(fd, buf, sizeof(buf)); + if (n > 0) { + auto lines = c.reader.feed(std::string_view(buf, static_cast(n))); + const bool overflowed = c.reader.overflowed(); + for (auto& line : lines) { + handle_line(c, line); + // handle_line may have replied with a fatal error and closed the + // connection (e.g. a protocol-major mismatch). Once that happens `c` + // is dangling — stop touching it. + if (conns_.find(fd) == conns_.end()) return; + } + if (overflowed) { + close_conn(fd); + return; + } + continue; + } + if (n == 0) { // peer closed + close_conn(fd); + return; + } + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + if (errno == EINTR) continue; + close_conn(fd); + return; + } + } +} + +void UdsServer::handle_line(Conn& c, const std::string& line) { + json req = json::parse(line, nullptr, /*allow_exceptions=*/false); + if (req.is_discarded()) { + queue_reply(c, rpc_error(nullptr, proto::ErrorCode::ParseError, "invalid JSON")); + return; + } + + const json id = req.is_object() && req.contains("id") ? req.at("id") : json(nullptr); + const std::string method = + req.is_object() && req.contains("method") && req.at("method").is_string() + ? req.at("method").get() + : std::string{}; + + if (!method.empty()) { + json reply; + if (handle_session_method(c, method, req, reply)) { + if (!reply.is_null()) queue_reply(c, reply); + return; + } + } + + // Everything else: the generated router. It returns a null json for a notification + // that needs no reply. + json reply = proto::dispatch(dispatcher_, proto::Transport::Uds, req); + if (!reply.is_null()) queue_reply(c, reply); +} + +bool UdsServer::handle_session_method(Conn& c, const std::string& method, const json& request, + json& reply) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params = request.contains("params") ? request.at("params") : json::object(); + + if (method == "session.hello") { + auto p = proto::parse(params, "params"); + if (!p) { + reply = rpc_error(id, proto::ErrorCode::InvalidParams, p.error().message, + json{{"path", p.error().path}}); + return true; + } + const int want = major_of(std::string(proto::kProtocolVersion)); + const int got = major_of(p->protocolVersion); + if (got != want) { + reply = rpc_error( + id, proto::ErrorCode::VersionMismatch, + "protocol major version mismatch: daemon speaks " + std::to_string(want) + + ".x, client speaks " + std::to_string(got < 0 ? 0 : got) + ".x", + json{{"expected", std::string(proto::kProtocolVersion)}, + {"actual", p->protocolVersion}}); + c.close_after_flush = true; // no method is served on a mismatched major + return true; + } + + c.hello_ok = true; + if (c.session_id.empty()) c.session_id = uuid4(); + + proto::SessionHelloResult r; + r.daemonVersion = std::string(velox::daemon::kDaemonVersion); + r.protocolVersion = std::string(proto::kProtocolVersion); + r.sessionId = c.session_id; + r.transport = proto::SessionHelloResultTransport::Uds; + reply = proto::make_result(id, r); + return true; + } + + if (method == "session.subscribe") { + auto p = proto::parse(params, "params"); + if (!p) { + reply = rpc_error(id, proto::ErrorCode::InvalidParams, p.error().message, + json{{"path", p.error().path}}); + return true; + } + // Event fan-out is not wired yet; accept the subscription and echo it back so a + // client can already register its interest without erroring. + proto::SessionSubscribeResult r; + r.ok = true; + for (const auto& ev : p->events) r.events.emplace_back(proto::to_string(ev)); + reply = proto::make_result(id, r); + return true; + } + + return false; // session.pair falls through to dispatch() -> -32003 on the Unix socket +} + +void UdsServer::queue_reply(Conn& c, const json& reply) { + c.outbuf += frame(reply.dump()); + if (c.outbuf.size() - c.out_off > kMaxOutBytes) { + close_conn(c.fd); + return; + } + flush(c); +} + +void UdsServer::flush(Conn& c) { + while (c.out_off < c.outbuf.size()) { + const ssize_t n = + ::write(c.fd, c.outbuf.data() + c.out_off, c.outbuf.size() - c.out_off); + if (n > 0) { + c.out_off += static_cast(n); + continue; + } + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + loop_.mod_fd(c.fd, kRead | kWrite); + return; + } + if (n < 0 && errno == EINTR) continue; + close_conn(c.fd); + return; + } + + c.outbuf.clear(); + c.out_off = 0; + if (c.close_after_flush) { + close_conn(c.fd); + return; + } + loop_.mod_fd(c.fd, kRead); +} + +void UdsServer::close_conn(int fd) { + if (const auto it = conns_.find(fd); it != conns_.end()) { + loop_.del_fd(fd); + ::close(fd); + conns_.erase(it); + } +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.hpp b/daemon/src/rpc/uds_server.hpp new file mode 100644 index 0000000..48cf389 --- /dev/null +++ b/daemon/src/rpc/uds_server.hpp @@ -0,0 +1,77 @@ +#pragma once + +// The Unix-domain-socket RPC listener: $XDG_RUNTIME_DIR/velox/velox.sock, mode 0600, +// SO_PEERCRED same-UID check (docs/01 §2, AGENT-DAEMON.md build step 1). NDJSON framing. +// Non-blocking throughout; every fd runs through the shared EventLoop so one slow client +// never stalls another. +// +// session.hello and session.subscribe are handled here because they are connection- and +// transport-stateful (protocol-major check, sessionId, per-connection subscription set). +// Every other method is routed through the generated velox::proto::dispatch(), which does +// the envelope, the -32003 privileged-transport refusal and the typed param parse. + +#include +#include +#include +#include +#include +#include + +#include + +#include "rpc/ndjson.hpp" +#include "velox_proto.hpp" + +namespace velox::daemon::rpc { + +class EventLoop; + +class UdsServer { +public: + UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, std::string socket_path); + ~UdsServer(); + + UdsServer(const UdsServer&) = delete; + UdsServer& operator=(const UdsServer&) = delete; + + // Create the socket, bind, chmod 0600, listen, and register with the loop. A stale + // socket file left by a crashed daemon is removed first. Returns a non-ok error_code + // (and changes nothing) on any failure. + std::error_code start(); + + const std::string& socket_path() const noexcept { return path_; } + std::size_t connection_count() const noexcept { return conns_.size(); } + +private: + struct Conn { + int fd; + FrameReader reader; + std::string outbuf; + std::size_t out_off = 0; // bytes of outbuf already written + bool close_after_flush = false; + bool hello_ok = false; + std::string session_id; + }; + + void on_listener_readable(); + void on_conn_event(int fd, unsigned events); + void handle_line(Conn& c, const std::string& line); + + // Returns true and fills `reply` if `method` is one this layer answers directly + // (session.hello / session.subscribe). Returns false to let dispatch() handle it. + bool handle_session_method(Conn& c, const std::string& method, const nlohmann::json& request, + nlohmann::json& reply); + + void queue_reply(Conn& c, const nlohmann::json& reply); + void flush(Conn& c); + void close_conn(int fd); + + EventLoop& loop_; + velox::proto::Dispatcher& dispatcher_; + std::string path_; + int listen_fd_ = -1; + bool bound_ = false; // path_ is ours to unlink on destruction + std::unordered_map> conns_; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/version.hpp b/daemon/src/version.hpp new file mode 100644 index 0000000..e8777d0 --- /dev/null +++ b/daemon/src/version.hpp @@ -0,0 +1,12 @@ +#pragma once + +#include + +namespace velox::daemon { + +// The daemon's own build version, reported in session.hello as daemonVersion. Distinct +// from the protocol version (velox::proto::kProtocolVersion), which is what the major +// compatibility check keys on. +inline constexpr std::string_view kDaemonVersion = "0.1.0"; + +} // namespace velox::daemon diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt new file mode 100644 index 0000000..04983f7 --- /dev/null +++ b/daemon/tests/CMakeLists.txt @@ -0,0 +1,14 @@ +# daemon unit + integration tests. Registered with ctest; run via `ctest --preset dev`. +# No external test framework — each file is a small self-checking binary (matches the +# lightweight style core/ uses, without depending on core's private test support). + +add_executable(veloxd_ndjson_test ndjson_test.cpp) +target_link_libraries(veloxd_ndjson_test PRIVATE veloxd_rpc) +target_compile_options(veloxd_ndjson_test PRIVATE -Wall -Wextra -Wpedantic -Werror) +add_test(NAME veloxd.ndjson COMMAND veloxd_ndjson_test) + +add_executable(veloxd_uds_roundtrip_test uds_roundtrip_test.cpp) +target_link_libraries(veloxd_uds_roundtrip_test PRIVATE veloxd_rpc) +target_compile_options(veloxd_uds_roundtrip_test PRIVATE -Wall -Wextra -Wpedantic -Werror) +add_test(NAME veloxd.uds_roundtrip COMMAND veloxd_uds_roundtrip_test) +set_tests_properties(veloxd.uds_roundtrip PROPERTIES TIMEOUT 30) diff --git a/daemon/tests/check.hpp b/daemon/tests/check.hpp new file mode 100644 index 0000000..5e5ce64 --- /dev/null +++ b/daemon/tests/check.hpp @@ -0,0 +1,53 @@ +#pragma once + +// Minimal test harness: CHECK accumulates failures, TEST_MAIN reports and sets the exit +// code. Deliberately tiny — the daemon does not pull in a test framework for this. + +#include +#include +#include + +namespace veloxd_test { + +inline std::vector& failures() { + static std::vector f; + return f; +} +inline int& checks() { + static int n = 0; + return n; +} + +} // namespace veloxd_test + +#define CHECK(cond) \ + do { \ + ++::veloxd_test::checks(); \ + if (!(cond)) { \ + ::veloxd_test::failures().push_back(std::string(__FILE__) + ":" + \ + std::to_string(__LINE__) + ": " + #cond); \ + } \ + } while (0) + +#define CHECK_EQ(a, b) \ + do { \ + ++::veloxd_test::checks(); \ + auto _va = (a); \ + auto _vb = (b); \ + if (!(_va == _vb)) { \ + ::veloxd_test::failures().push_back(std::string(__FILE__) + ":" + \ + std::to_string(__LINE__) + ": " + #a + \ + " == " + #b); \ + } \ + } while (0) + +#define TEST_MAIN() \ + int main() { \ + run(); \ + for (const auto& f : ::veloxd_test::failures()) std::printf("FAIL %s\n", f.c_str()); \ + std::printf("%d/%d checks passed\n", \ + ::veloxd_test::checks() - \ + static_cast(::veloxd_test::failures().size()), \ + ::veloxd_test::checks()); \ + return ::veloxd_test::failures().empty() ? 0 : 1; \ + } diff --git a/daemon/tests/ndjson_test.cpp b/daemon/tests/ndjson_test.cpp new file mode 100644 index 0000000..ef6b1c1 --- /dev/null +++ b/daemon/tests/ndjson_test.cpp @@ -0,0 +1,64 @@ +#include "rpc/ndjson.hpp" + +#include + +#include "check.hpp" + +using velox::daemon::rpc::FrameReader; +using velox::daemon::rpc::frame; + +void run() { + // One chunk, one frame. + { + FrameReader r; + auto f = r.feed("{\"a\":1}\n"); + CHECK_EQ(f.size(), 1u); + CHECK_EQ(f.at(0), std::string("{\"a\":1}")); + CHECK_EQ(r.buffered(), 0u); + } + + // A frame split across two feeds is delivered only when the newline arrives. + { + FrameReader r; + CHECK_EQ(r.feed("{\"a\":").size(), 0u); + auto f = r.feed("1}\n"); + CHECK_EQ(f.size(), 1u); + CHECK_EQ(f.at(0), std::string("{\"a\":1}")); + } + + // Several frames in one chunk, plus a partial tail held back. + { + FrameReader r; + auto f = r.feed("1\n2\n3\n4"); + CHECK_EQ(f.size(), 3u); + CHECK_EQ(f.at(0), std::string("1")); + CHECK_EQ(f.at(2), std::string("3")); + CHECK_EQ(r.buffered(), 1u); + auto g = r.feed("\n"); + CHECK_EQ(g.size(), 1u); + CHECK_EQ(g.at(0), std::string("4")); + } + + // CRLF terminator is trimmed; blank lines are skipped. + { + FrameReader r; + auto f = r.feed("x\r\n\r\n\ny\r\n"); + CHECK_EQ(f.size(), 2u); + CHECK_EQ(f.at(0), std::string("x")); + CHECK_EQ(f.at(1), std::string("y")); + } + + // Overflow latches once the unframed tail passes the cap. + { + FrameReader r; + CHECK(!r.overflowed()); + std::string big(9u * 1024 * 1024, 'a'); // no newline + r.feed(big); + CHECK(r.overflowed()); + } + + // frame() appends exactly one newline. + CHECK_EQ(frame("hello"), std::string("hello\n")); +} + +TEST_MAIN() diff --git a/daemon/tests/uds_roundtrip_test.cpp b/daemon/tests/uds_roundtrip_test.cpp new file mode 100644 index 0000000..39f8ea9 --- /dev/null +++ b/daemon/tests/uds_roundtrip_test.cpp @@ -0,0 +1,195 @@ +// Integration: a real UdsServer on a temp socket, a real client socket, NDJSON round trips. +// Proves the transport, the generated dispatch() wiring, and the session-layer handling. + +#include +#include +#include + +#include +#include +#include +#include + +#include + +#include "check.hpp" +#include "rpc/dispatcher.hpp" +#include "rpc/event_loop.hpp" +#include "rpc/ndjson.hpp" +#include "rpc/uds_server.hpp" +#include "velox_proto.hpp" + +using nlohmann::json; +namespace rpc = velox::daemon::rpc; + +namespace { + +std::string make_temp_socket_path() { + char tmpl[] = "/tmp/veloxd-test-XXXXXX"; + const char* dir = ::mkdtemp(tmpl); + return std::string(dir ? dir : "/tmp") + "/velox.sock"; +} + +int connect_client(const std::string& path) { + const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + std::memcpy(addr.sun_path, path.c_str(), path.size()); + if (::connect(fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + ::close(fd); + return -1; + } + return fd; +} + +// Send one framed request, read one framed reply (blocking client; the server is async). +json call(int fd, const json& request) { + const std::string out = rpc::frame(request.dump()); + if (::write(fd, out.data(), out.size()) != static_cast(out.size())) return {}; + + std::string buf; + char chunk[4096]; + for (;;) { + const ssize_t n = ::read(fd, chunk, sizeof(chunk)); + if (n <= 0) return {}; + buf.append(chunk, static_cast(n)); + if (const auto nl = buf.find('\n'); nl != std::string::npos) + return json::parse(buf.substr(0, nl), nullptr, false); + } +} + +} // namespace + +void run() { + const std::string sock = make_temp_socket_path(); + + rpc::EventLoop loop; + rpc::VeloxDispatcher dispatcher; + rpc::UdsServer server(loop, dispatcher, sock); + const auto ec = server.start(); + CHECK(!ec); + if (ec) return; + + std::thread loop_thread([&loop] { loop.run(); }); + + // --- session.hello, matching major -> a real SessionHelloResult ------------------- + { + const int c = connect_client(sock); + CHECK(c >= 0); + const json reply = call(c, {{"jsonrpc", "2.0"}, + {"id", 1}, + {"method", "session.hello"}, + {"params", + {{"clientType", "test"}, + {"clientName", "roundtrip"}, + {"protocolVersion", std::string(velox::proto::kProtocolVersion)}}}}); + CHECK(reply.contains("result")); + CHECK_EQ(reply["id"].get(), 1); + CHECK_EQ(reply["result"]["protocolVersion"].get(), + std::string(velox::proto::kProtocolVersion)); + CHECK_EQ(reply["result"]["transport"].get(), std::string("uds")); + CHECK(!reply["result"]["sessionId"].get().empty()); + ::close(c); + } + + // --- session.hello, wrong major -> -32001, connection closed after the reply ------ + { + const int c = connect_client(sock); + const json reply = call(c, {{"jsonrpc", "2.0"}, + {"id", 2}, + {"method", "session.hello"}, + {"params", + {{"clientType", "gui"}, + {"clientName", "from the future"}, + {"protocolVersion", "2.0.0"}}}}); + CHECK(reply.contains("error")); + CHECK_EQ(reply["error"]["code"].get(), -32001); + CHECK_EQ(reply["error"]["data"]["actual"].get(), std::string("2.0.0")); + ::close(c); + } + + // --- download.list -> an empty table (dispatcher answers this one for real) ------- + { + const int c = connect_client(sock); + const json reply = + call(c, {{"jsonrpc", "2.0"}, {"id", 3}, {"method", "download.list"}, {"params", json::object()}}); + CHECK(reply.contains("result")); + CHECK_EQ(reply["result"]["total"].get(), 0); + CHECK(reply["result"]["items"].is_array()); + CHECK_EQ(reply["result"]["items"].size(), 0u); + ::close(c); + } + + // --- an unknown method -> -32601 ------------------------------------------------ + { + const int c = connect_client(sock); + const json reply = + call(c, {{"jsonrpc", "2.0"}, {"id", 4}, {"method", "no.such.method"}, {"params", json::object()}}); + CHECK(reply.contains("error")); + CHECK_EQ(reply["error"]["code"].get(), -32601); + ::close(c); + } + + // --- malformed JSON -> -32700, id null ---------------------------------------- + { + const int c = connect_client(sock); + const std::string bad = "{ this is not json )\n"; + CHECK(::write(c, bad.data(), bad.size()) == static_cast(bad.size())); + std::string buf; + char chunk[1024]; + const ssize_t n = ::read(c, chunk, sizeof(chunk)); + CHECK(n > 0); + if (n > 0) { + buf.assign(chunk, static_cast(n)); + const json reply = json::parse(buf.substr(0, buf.find('\n')), nullptr, false); + CHECK_EQ(reply["error"]["code"].get(), -32700); + CHECK(reply["id"].is_null()); + } + ::close(c); + } + + // --- download.get -> -32603 for now: documents the P1 codegen gap --------------- + // (proto-requests-m1.md P1: handlers cannot yet return -32010. When P1 lands this + // check flips to -32010 and is the regression guard for it.) + { + const int c = connect_client(sock); + const json reply = call(c, {{"jsonrpc", "2.0"}, + {"id", 6}, + {"method", "download.get"}, + {"params", {{"taskId", "00000000-0000-4000-8000-000000000000"}}}}); + CHECK(reply.contains("error")); + CHECK_EQ(reply["error"]["code"].get(), -32603); + ::close(c); + } + + // --- two requests in one write, pipelined on one connection -------------------- + { + const int c = connect_client(sock); + std::string out = rpc::frame(json({{"jsonrpc", "2.0"}, {"id", 7}, {"method", "download.list"}, {"params", json::object()}}).dump()); + out += rpc::frame(json({{"jsonrpc", "2.0"}, {"id", 8}, {"method", "download.list"}, {"params", json::object()}}).dump()); + CHECK(::write(c, out.data(), out.size()) == static_cast(out.size())); + std::string buf; + char chunk[4096]; + int seen = 0; + while (seen < 2) { + const ssize_t n = ::read(c, chunk, sizeof(chunk)); + if (n <= 0) break; + buf.append(chunk, static_cast(n)); + std::size_t nl; + while ((nl = buf.find('\n')) != std::string::npos) { + const json reply = json::parse(buf.substr(0, nl), nullptr, false); + CHECK(reply.contains("result")); + ++seen; + buf.erase(0, nl + 1); + } + } + CHECK_EQ(seen, 2); + ::close(c); + } + + loop.stop(); + loop_thread.join(); + ::unlink(sock.c_str()); +} + +TEST_MAIN()