diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index a42e211..fe73e4d 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -74,7 +74,6 @@ target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------ add_library(veloxd_rpc STATIC src/rpc/runtime_dir.cpp - src/rpc/single_instance.cpp src/rpc/systemd_activation.cpp src/rpc/event_loop.cpp src/rpc/event_hub.cpp @@ -84,6 +83,14 @@ add_library(veloxd_rpc STATIC src/rpc/ws_server.cpp src/rpc/pairing.cpp src/rpc/dispatcher.cpp + # Per-OS backends behind daemon/src/rpc/platform/*.hpp (docs/adr/0020, + # docs/08-porting.md). Only the Linux backend exists (Phase 0); lane PORT adds + # cmake/platform.cmake + platform/macos/ in Phase 1 and this hardcoded list becomes + # a velox_platform_sources() call. + src/rpc/platform/linux/wakeup.cpp + src/rpc/platform/linux/peer_id.cpp + src/rpc/platform/linux/instance_lock.cpp + src/rpc/platform/linux/runtime_paths.cpp ) add_library(velox::daemon_rpc ALIAS veloxd_rpc) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 6b2ca6e..0b76a19 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -27,4 +27,5 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | ~~D5~~ | **Mostly closed** — `rpc/event_hub` fans out per-subscription; `session.subscribe` on both transports registers/updates/tears down a real subscription; `Scheduler::transition()` publishes `event.task.state` (with `previousState`) on every state change, scheduler-driven or engine-reported; `dispatcher::on_download_add` publishes `event.task.added`; a 250 ms timer batches `Scheduler::progress_snapshot()` into one `event.task.progress` array per AGENT-DAEMON.md item 5 / the schema's `x-maxRateHz: 4`. Verified live end to end. | — | `event.task.removed` has no source yet (`download.remove` is D3); `event.speed.global`, `event.notify`, `event.auth.required`, `event.settings.changed`, `event.grabber.progress` are unpublished — each lands with its owning handler | as each owning D3 handler lands | | ~~D6~~ | **Closed** — engine numbers now reach the store: `Scheduler::tick()` probes (`EnginePort::probe`) before every `start()`, persisting `sizeBytes`/`resumable`/validators via `Tasks::set_probe_result` before a byte moves; `Scheduler::persist_progress()` (called from `progress_snapshot()` *and* once more from `on_engine_state` right before `release()`/unmap on every terminal transition) writes `downloadedBytes`/`speedBps`/`segments`/`segmentDetail` from the engine's `Progress`, so a task that finishes between two 250 ms ticks (the common case for anything small or fast) still leaves real numbers instead of the pre-persistence defaults. `TaskSummary.segments` is sourced from `segments.size()` when the task has any (matching what actually lands in `segmentDetail`, per the schema's "exactly `segments` entries"), falling back to the engine's `effective_segments` (budget slots *held*, not necessarily physical range count — see `core/include/vdm/task/download.hpp`'s `Progress` comment) only pre-segmentation. `Tasks::set_final_bytes` tops up `on_finished`'s byte count as a last-resort backstop. Migration `0002` adds `speed_bps` to both `tasks` and `segments`, and fixes `segments.state`'s CHECK to include `'pending'` (0001 omitted it, so a pre-connect snapshot could never be written). Verified against real `veloxd` + `tools/testserver` (not just unit tests): `download.list`/`download.get` correct immediately after completion and after a daemon restart. | `sched/scheduler.{cpp,hpp}`, `store/{tasks,segments}.{cpp,hpp}`, `store/migrations/0002_*.sql` | — | done | | ~~D11~~ | **Closed — build order items 7 (the systemd half) and 9: `velox-nmhost`, socket activation, the systemd user units, and `velox(1)`.** `nmhost/src/main.cpp` (185 lines): a `poll()`-driven byte pump between Firefox's native-messaging framing on stdio (4-byte native-byte-order length prefix) and `veloxd`'s own NDJSON framing on the Unix socket — reframes each direction, no JSON parsing, no retry/backoff, exits the moment either side closes. Deliberately dependency-free (no `veloxd_*` library, no `nlohmann_json`) since it runs unconfined outside Firefox's sandbox whatever the packaging format. Two real bugs found and fixed while getting the integration test to actually pass rather than hang: (1) never set the pumped fds non-blocking, so the "drain what's available" read loop blocked on its own second `read()` instead of returning to `poll()`; (2) stdin and stdout are two different descriptors (0 and 1), not one — an early draft polled `POLLOUT` on fd 0, which is opened read-only, so EOF/writability were never both observable through the same `pollfd` entry. Both are exactly the class of bug a "trivial pump" invites and unit tests over the real binary (not just its helper functions) exist specifically to catch. `packaging/nativehost/com.velox.host.json` + its own `README.md` supersede `AGENT-DAEMON.md`'s stale "four locations" (spike S1 / ADR 0003 found only three are real — the fourth, `~/snap/firefox/common/.mozilla/...`, is not read by snap Firefox at all) and spell out the per-user-manifest / postinst implication for PKG/QA. `EnginePort`-style: `rpc/systemd_activation.cpp` is a from-scratch `sd_listen_fds()` (env vars only, no `libsystemd` link — `LISTEN_PID`/`LISTEN_FDS`, fd 3) that `UdsServer::start()` checks first, skipping its own create/bind/chmod/listen when systemd already bound the socket; `packaging/systemd/velox.socket` + `velox.service` are the unit pair, verified both by `systemd-analyze verify` and by an actual fork/dup2/execve simulation of the activation handshake (a real `session.hello` round-tripped over the handed-off fd with no `bind()` ever called inside the daemon for that run). `velox.service` deliberately skips `ProtectSystem=`/`ProtectHome=`/`ReadWritePaths=` — `saveTo.allowedRoots` is user-configurable to anywhere on the filesystem, and a sandbox here would turn a legitimately-configured save location into an opaque `EROFS`/`EACCES` instead of the daemon's own clear `-32011`. `cli/man/velox.1` documents the CLI as it actually exists today (`add`/`ls`/`pause`/`resume`/`rm`, `--json`, the three-tier `queue`/`settings` subcommands `AGENT-DAEMON.md` build step 8 originally sketched are not implemented in `cli/src/main.cpp` yet, so the page doesn't claim they are) — checked warning-free with `groff -mandoc -ww -z`. | `nmhost/{CMakeLists.txt,src/main.cpp,tests/}`, `daemon/src/rpc/{systemd_activation.{hpp,cpp},uds_server.cpp}`, `packaging/{nativehost,systemd}/`, `cli/man/velox.1` | — | done | +| ~~D12~~ | **Closed — ADR 0020 Phase 0 for `daemon/`.** Pure refactor, zero behaviour change: the four Linux-only surfaces docs/08-porting.md lists for this lane now sit behind seams in `daemon/src/rpc/platform/{wakeup,peer_id,instance_lock,runtime_paths}.hpp`, with today's syscalls moved unchanged into `platform/linux/*.cpp` (`eventfd` -> `platform::Wakeup`, `SO_PEERCRED`/`struct ucred` -> `platform::peer_of` — same-UID check preserved verbatim as the security property, not touched — abstract-namespace socket -> `platform::acquire_instance_lock`, XDG env lookups -> `platform::runtime_base_dir`/`data_base_dir`, with the 0700-and-owned enforcement itself staying portable POSIX logic in `rpc/runtime_dir.cpp` rather than moving). No fifth seam for `timerfd`: both daemon timers (1 s scheduler tick, 250 ms progress batch) now go through a new portable `EventLoop::add_timer`, which folds the next timer deadline into the existing `poll()` timeout instead of a platform timer fd — exactly the case docs/08-porting.md calls out as not needing a backend ("the loop already has a deadline set"), so it stays in `rpc/event_loop.{hpp,cpp}` untouched by lane PORT. `daemon/CMakeLists.txt` lists the four `platform/linux/*.cpp` files directly (no `cmake/platform.cmake` yet — that and `platform/macos/` are PORT's, per the CLAUDE.md lane table added alongside ADR 0020); PORT's Phase 1 replaces the hardcoded list with `velox_platform_sources()`. Old `rpc/single_instance.{hpp,cpp}` deleted (folded into the new seam); `single_instance_test.cpp` retargeted at `rpc/platform/instance_lock.hpp` with no behaviour change. Full suite 59/59 green; `git diff` outside `platform/linux/` is moves/call-site updates only, no `#ifdef` anywhere in `daemon/`. | `rpc/platform/{wakeup,peer_id,instance_lock,runtime_paths}.hpp`, `rpc/platform/linux/*.cpp`, `rpc/event_loop.{hpp,cpp}`, `rpc/{uds_server,runtime_dir}.cpp`, `main.cpp`, `CMakeLists.txt` | — | done | | — | ~~Observed, not fixed (CORE, not this lane)~~ — **routed to CORE by the user.** `vdm::task::Progress.speed_bps` reads back as `0` for the whole lifetime of a live, real (non-fake) throttled download, despite `downloadedBytes` visibly advancing between polls — `core/src/task/download_task.cpp`'s per-worker EWMA never seems to produce a nonzero aggregate in this build. DAEMON passes `EnginePort::progress()`'s `speed_bps` straight through (`Scheduler::persist_progress`); nothing in this lane drops it. Still reproduces in the D4b live checks above (0 throughout a paused/resumed/cancelled transfer whose `downloadedBytes` visibly moved) — not re-filed, since it's already CORE's. | diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index 7b1160d..f3f1f09 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -5,6 +5,7 @@ // (AGENT-DAEMON.md build order, steps 1 and 3). The scheduler and the engine link land // next. +#include #include #include #include @@ -15,16 +16,14 @@ #include #include -#include - #include #include "rpc/dispatcher.hpp" #include "rpc/event_hub.hpp" #include "rpc/event_loop.hpp" #include "rpc/pairing.hpp" +#include "rpc/platform/instance_lock.hpp" #include "rpc/runtime_dir.hpp" -#include "rpc/single_instance.hpp" #include "rpc/uds_server.hpp" #include "rpc/ws_server.hpp" #include "sched/engine_port_core.hpp" @@ -56,7 +55,7 @@ int main() { return 1; } - const int lock_fd = velox::daemon::rpc::acquire_single_instance_lock(rt.path); + const int lock_fd = velox::daemon::rpc::platform::acquire_instance_lock(rt.path); if (lock_fd < 0) { std::cerr << "veloxd: another instance is already running for this runtime " "directory (" << rt.path << ")\n"; @@ -109,57 +108,39 @@ int main() { }); // A 1 s timer re-runs the scheduler so schedule windows opening/closing and any - // missed nudge are picked up. Registered on the loop, no extra thread. - const int tick_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); - if (tick_fd >= 0) { - itimerspec spec{}; - spec.it_value.tv_sec = 1; - spec.it_interval.tv_sec = 1; - ::timerfd_settime(tick_fd, 0, &spec, nullptr); - loop.add_fd(tick_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) { - std::uint64_t ticks = 0; - [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks)); - (void)scheduler.tick(); - }); - } + // missed nudge are picked up. Expressed as a poll() timeout behind EventLoop rather + // than a timerfd (docs/08-porting.md: "the loop already has a deadline set" — no + // per-OS backend needed here, unlike the other three daemon/rpc seams). + loop.add_timer(std::chrono::seconds(1), [&] { (void)scheduler.tick(); }); // event.task.progress: one array message at <=4 Hz (schema x-maxRateHz), never one // notification per task (AGENT-DAEMON.md item 5). 250 ms keeps every active task's // segment bar under 4 Hz without depending on how many tasks are running. - const int progress_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC); - if (progress_fd >= 0) { - itimerspec spec{}; - spec.it_value.tv_nsec = 250'000'000; - spec.it_interval.tv_nsec = 250'000'000; - ::timerfd_settime(progress_fd, 0, &spec, nullptr); - loop.add_fd(progress_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) { - std::uint64_t ticks = 0; - [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks)); - const auto rows = scheduler.progress_snapshot(); - if (rows.empty()) return; + loop.add_timer(std::chrono::milliseconds(250), [&] { + const auto rows = scheduler.progress_snapshot(); + if (rows.empty()) return; - nlohmann::json tasks_json = nlohmann::json::array(); - for (const auto& r : rows) { - nlohmann::json t{{"taskId", r.task_id}, - {"downloadedBytes", r.downloaded_bytes}, - {"speedBps", r.speed_bps}}; - t["etaSeconds"] = r.eta_seconds ? nlohmann::json(*r.eta_seconds) : nlohmann::json(nullptr); - if (!r.segments.empty()) { - nlohmann::json segs = nlohmann::json::array(); - for (const auto& s : r.segments) - segs.push_back({{"index", s.index}, - {"downloadedBytes", s.downloaded_bytes}, - {"speedBps", s.speed_bps}}); - t["segments"] = std::move(segs); - } - tasks_json.push_back(std::move(t)); + nlohmann::json tasks_json = nlohmann::json::array(); + for (const auto& r : rows) { + nlohmann::json t{{"taskId", r.task_id}, + {"downloadedBytes", r.downloaded_bytes}, + {"speedBps", r.speed_bps}}; + t["etaSeconds"] = r.eta_seconds ? nlohmann::json(*r.eta_seconds) : nlohmann::json(nullptr); + if (!r.segments.empty()) { + nlohmann::json segs = nlohmann::json::array(); + for (const auto& s : r.segments) + segs.push_back({{"index", s.index}, + {"downloadedBytes", s.downloaded_bytes}, + {"speedBps", s.speed_bps}}); + t["segments"] = std::move(segs); } - const nlohmann::json params{{"tasks", std::move(tasks_json)}, - {"at", velox::daemon::now_iso()}}; - hub.publish(velox::proto::Event::TaskProgress, - velox::proto::make_notification(velox::proto::Event::TaskProgress, params)); - }); - } + tasks_json.push_back(std::move(t)); + } + const nlohmann::json params{{"tasks", std::move(tasks_json)}, + {"at", velox::daemon::now_iso()}}; + hub.publish(velox::proto::Event::TaskProgress, + velox::proto::make_notification(velox::proto::Event::TaskProgress, params)); + }); velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path(), &scheduler); if (const auto ec = uds.start()) { @@ -186,14 +167,6 @@ int main() { loop.run(); std::cout << "veloxd: shutting down\n"; - if (tick_fd >= 0) { - loop.del_fd(tick_fd); - ::close(tick_fd); - } - if (progress_fd >= 0) { - loop.del_fd(progress_fd); - ::close(progress_fd); - } g_loop = nullptr; ::close(lock_fd); return 0; diff --git a/daemon/src/rpc/event_loop.cpp b/daemon/src/rpc/event_loop.cpp index 7c15807..19df529 100644 --- a/daemon/src/rpc/event_loop.cpp +++ b/daemon/src/rpc/event_loop.cpp @@ -1,9 +1,8 @@ #include "rpc/event_loop.hpp" #include -#include -#include +#include #include #include #include @@ -12,14 +11,10 @@ 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(); }}); + fds_.emplace(wakeup_.pollfd(), Entry{kRead, [this](int, unsigned) { wakeup_.drain(); }}); } -EventLoop::~EventLoop() { - if (wake_fd_ >= 0) ::close(wake_fd_); -} +EventLoop::~EventLoop() = default; void EventLoop::add_fd(int fd, unsigned interest, Callback cb) { fds_[fd] = Entry{interest, std::move(cb)}; @@ -30,15 +25,11 @@ void EventLoop::mod_fd(int fd, unsigned interest) { } void EventLoop::del_fd(int fd) { - if (fd == wake_fd_) return; // internal, never removed + if (fd == wakeup_.pollfd()) 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::wake() noexcept { wakeup_.signal(); } void EventLoop::stop() noexcept { stop_requested_ = true; @@ -62,9 +53,39 @@ void EventLoop::drain_posts() { for (auto& fn : batch) fn(); } -void EventLoop::drain_wakeup() noexcept { - std::uint64_t sink = 0; - while (::read(wake_fd_, &sink, sizeof(sink)) > 0) { +EventLoop::TimerId EventLoop::add_timer(std::chrono::milliseconds interval, + std::function cb) { + const TimerId id = next_timer_id_++; + timers_.emplace(id, Timer{std::chrono::steady_clock::now() + interval, interval, + std::move(cb)}); + wake(); // the loop may already be blocked on a longer timeout + return id; +} + +void EventLoop::remove_timer(TimerId id) { timers_.erase(id); } + +int EventLoop::next_timeout_ms() const { + if (timers_.empty()) return -1; + auto soonest = timers_.begin()->second.next; + for (const auto& [id, t] : timers_) soonest = std::min(soonest, t.next); + const auto now = std::chrono::steady_clock::now(); + const auto ms = std::chrono::duration_cast(soonest - now); + return ms.count() > 0 ? static_cast(ms.count()) : 0; +} + +void EventLoop::run_due_timers() { + const auto now = std::chrono::steady_clock::now(); + // Snapshot ids first: a callback may add/remove timers, which would invalidate + // iteration over timers_ directly. + std::vector due; + for (auto& [id, t] : timers_) { + if (t.next <= now) due.push_back(id); + } + for (const TimerId id : due) { + const auto it = timers_.find(id); + if (it == timers_.end()) continue; // removed by an earlier callback this pass + it->second.next = now + it->second.interval; + it->second.cb(); } } @@ -83,18 +104,22 @@ void EventLoop::run() { short ev = 0; if (e.interest & kRead) ev |= POLLIN; if (e.interest & kWrite) ev |= POLLOUT; - if (ev == 0 && fd != wake_fd_) continue; + if (ev == 0 && fd != wakeup_.pollfd()) continue; pollfd p{}; p.fd = fd; p.events = ev; pfds.push_back(p); } - const int rc = ::poll(pfds.data(), pfds.size(), -1); + const int rc = ::poll(pfds.data(), pfds.size(), next_timeout_ms()); if (rc < 0) { if (errno == EINTR) continue; throw std::runtime_error("poll() failed"); } + + drain_posts(); + run_due_timers(); + if (rc == 0) continue; // Snapshot the fds that fired before invoking any callback: a callback may erase @@ -104,8 +129,6 @@ void EventLoop::run() { if (p.revents != 0) fired.push_back(p.fd); } - drain_posts(); - for (const int fd : fired) { const auto it = fds_.find(fd); if (it == fds_.end()) continue; // removed by an earlier callback this pass diff --git a/daemon/src/rpc/event_loop.hpp b/daemon/src/rpc/event_loop.hpp index e6d80f5..882cebc 100644 --- a/daemon/src/rpc/event_loop.hpp +++ b/daemon/src/rpc/event_loop.hpp @@ -10,12 +10,15 @@ // from any thread or a signal handler — they only write() a byte to an internal eventfd. #include +#include #include #include #include #include #include +#include "rpc/platform/wakeup.hpp" + namespace velox::daemon::rpc { enum Interest : unsigned { @@ -58,22 +61,43 @@ public: // marshal an engine-thread callback back onto the RPC loop. void post(std::function fn); + using TimerId = std::uint64_t; + + // Fire `cb` roughly every `interval` for as long as the loop runs, expressed as a + // poll(2) timeout rather than a platform timer fd (docs/08-porting.md: "the loop + // already has a deadline set" — no per-OS backend needed for this one). Not + // reentrant-safe to call from inside a timer callback other than the one running. + TimerId add_timer(std::chrono::milliseconds interval, std::function cb); + void remove_timer(TimerId id); + private: struct Entry { unsigned interest; Callback cb; }; - void drain_wakeup() noexcept; - void drain_posts(); + struct Timer { + std::chrono::steady_clock::time_point next; + std::chrono::milliseconds interval; + std::function cb; + }; - int wake_fd_; // eventfd, always registered + void drain_posts(); + // Milliseconds until the next timer is due, or -1 if there are none (poll()'s "block + // forever" convention). + int next_timeout_ms() const; + void run_due_timers(); + + platform::Wakeup wakeup_; bool running_ = false; std::atomic stop_requested_ = false; // set from stop(), read by run() std::unordered_map fds_; std::mutex post_mu_; std::vector> posts_; + + TimerId next_timer_id_ = 1; + std::unordered_map timers_; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/platform/instance_lock.hpp b/daemon/src/rpc/platform/instance_lock.hpp new file mode 100644 index 0000000..de0382e --- /dev/null +++ b/daemon/src/rpc/platform/instance_lock.hpp @@ -0,0 +1,20 @@ +#pragma once + +// Single-instance guard, keyed by the resolved runtime directory (see rpc/runtime_dir.hpp) +// so isolated instances pointed at different runtime dirs never contend (docs/01 §2). The +// mechanism is Linux's abstract-namespace Unix socket; macOS has no abstract namespace and +// uses a real socket file plus flock() instead, which must unlink a stale socket left by a +// crashed process (docs/08-porting.md "API mapping" — the abstract version got that for +// free from the kernel). + +#include +#include + +namespace velox::daemon::rpc::platform { + +// Returns the held fd (kept open for the process lifetime; closing it releases the lock) +// or -1 if another process already holds the lock for this exact `runtime_dir`, or on any +// other error acquiring it. +int acquire_instance_lock(const std::string& runtime_dir); + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/single_instance.cpp b/daemon/src/rpc/platform/linux/instance_lock.cpp similarity index 85% rename from daemon/src/rpc/single_instance.cpp rename to daemon/src/rpc/platform/linux/instance_lock.cpp index 5385c96..50d0654 100644 --- a/daemon/src/rpc/single_instance.cpp +++ b/daemon/src/rpc/platform/linux/instance_lock.cpp @@ -1,4 +1,4 @@ -#include "rpc/single_instance.hpp" +#include "rpc/platform/instance_lock.hpp" #include #include @@ -9,9 +9,9 @@ #include "util/crypto.hpp" -namespace velox::daemon::rpc { +namespace velox::daemon::rpc::platform { -int acquire_single_instance_lock(const std::string& runtime_dir) { +int acquire_instance_lock(const std::string& runtime_dir) { const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); if (fd < 0) return -1; @@ -35,4 +35,4 @@ int acquire_single_instance_lock(const std::string& runtime_dir) { return fd; } -} // namespace velox::daemon::rpc +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/linux/peer_id.cpp b/daemon/src/rpc/platform/linux/peer_id.cpp new file mode 100644 index 0000000..a52f39a --- /dev/null +++ b/daemon/src/rpc/platform/linux/peer_id.cpp @@ -0,0 +1,17 @@ +#include "rpc/platform/peer_id.hpp" + +#include + +namespace velox::daemon::rpc::platform { + +std::error_code peer_of(int fd, PeerId& out) { + ucred cred{}; + socklen_t len = sizeof(cred); + if (::getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &cred, &len) != 0) { + return std::error_code(errno, std::generic_category()); + } + out.uid = cred.uid; + return {}; +} + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/linux/runtime_paths.cpp b/daemon/src/rpc/platform/linux/runtime_paths.cpp new file mode 100644 index 0000000..180e3ba --- /dev/null +++ b/daemon/src/rpc/platform/linux/runtime_paths.cpp @@ -0,0 +1,44 @@ +#include "rpc/platform/runtime_paths.hpp" + +#include +#include +#include + +#include +#include + +namespace velox::daemon::rpc::platform { + +namespace { +std::error_code errc(int e) { return std::error_code(e, std::generic_category()); } +} // namespace + +std::error_code runtime_base_dir(std::string& out) { + if (const char* xdg = ::getenv("XDG_RUNTIME_DIR"); xdg != nullptr && xdg[0] != '\0') { + out = xdg; + return {}; + } + const std::string 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/: refuse rather than pick an insecure + // fallback. The caller surfaces this as "cannot start". + return errc(ENOENT); + } + out = base; + return {}; +} + +std::error_code data_base_dir(std::string& out) { + if (const char* xdg = ::getenv("XDG_DATA_HOME"); xdg != nullptr && xdg[0] != '\0') { + out = xdg; + return {}; + } + if (const char* home = ::getenv("HOME"); home != nullptr && home[0] != '\0') { + out = std::string(home) + "/.local/share"; + return {}; + } + return errc(ENOENT); +} + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/linux/wakeup.cpp b/daemon/src/rpc/platform/linux/wakeup.cpp new file mode 100644 index 0000000..eb3350e --- /dev/null +++ b/daemon/src/rpc/platform/linux/wakeup.cpp @@ -0,0 +1,32 @@ +#include "rpc/platform/wakeup.hpp" + +#include +#include + +#include +#include + +namespace velox::daemon::rpc::platform { + +Wakeup::Wakeup() { + fd_ = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + if (fd_ < 0) throw std::runtime_error("eventfd() failed"); +} + +Wakeup::~Wakeup() { + if (fd_ >= 0) ::close(fd_); +} + +void Wakeup::signal() 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(fd_, &one, sizeof(one)); +} + +void Wakeup::drain() noexcept { + std::uint64_t sink = 0; + while (::read(fd_, &sink, sizeof(sink)) > 0) { + } +} + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/peer_id.hpp b/daemon/src/rpc/platform/peer_id.hpp new file mode 100644 index 0000000..7fcb004 --- /dev/null +++ b/daemon/src/rpc/platform/peer_id.hpp @@ -0,0 +1,22 @@ +#pragma once + +// Identifies the process on the other end of a connected Unix-domain socket, for the +// same-UID check that is the Unix transport's authorization boundary (docs/01 §2, +// CLAUDE.md §4). `SO_PEERCRED`/`struct ucred` is Linux-only; macOS has `getpeereid`, +// Windows named pipes carry a token instead (docs/08-porting.md "The seams" / +// "API mapping"). The same-UID check itself is the security property and must not change +// per-OS (docs/adr/0020 decision 2). + +#include + +namespace velox::daemon::rpc::platform { + +struct PeerId { + unsigned int uid = 0; +}; + +// On success, fills `out` with the peer's identity of the already-connected `fd`. On +// failure, `out` is untouched and the error_code explains why (matches errno on Linux). +std::error_code peer_of(int fd, PeerId& out); + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/runtime_paths.hpp b/daemon/src/rpc/platform/runtime_paths.hpp new file mode 100644 index 0000000..f58cc9b --- /dev/null +++ b/daemon/src/rpc/platform/runtime_paths.hpp @@ -0,0 +1,22 @@ +#pragma once + +// Where the OS wants ephemeral runtime state and persistent user data to live, before +// velox appends its own "/velox" subdirectory and applies the shared 0700-and-owned check +// (rpc/runtime_dir.cpp — that enforcement is portable POSIX logic and stays there; only +// "which base directory" is per-OS). Linux: XDG. macOS: $TMPDIR (runtime) and +// ~/Library/Application Support (data) — see docs/08-porting.md "API mapping". + +#include +#include + +namespace velox::daemon::rpc::platform { + +// The base directory ephemeral runtime state (sockets, lock files) should live under, +// e.g. "$XDG_RUNTIME_DIR" or "/run/user/" on Linux. No trailing slash. +std::error_code runtime_base_dir(std::string& out); + +// The base directory persistent user data should live under, e.g. "$XDG_DATA_HOME" or +// "~/.local/share" on Linux. No trailing slash. +std::error_code data_base_dir(std::string& out); + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/platform/wakeup.hpp b/daemon/src/rpc/platform/wakeup.hpp new file mode 100644 index 0000000..2985cb7 --- /dev/null +++ b/daemon/src/rpc/platform/wakeup.hpp @@ -0,0 +1,36 @@ +#pragma once + +// The event-loop wakeup primitive (docs/adr/0020, docs/08-porting.md). EventLoop uses this +// to interrupt a blocked poll() from another thread or a signal handler — the eventfd +// mechanics themselves are Linux-only; every other OS backend just needs something +// poll()-able that signal()/drain() can drive the same way (docs/08 §"The seams": a +// self-pipe on macOS, an event object on Windows). +// +// One implementation file per OS under platform//wakeup.cpp; this header carries no +// OS types and no #ifdef (ADR 0020 decision 1). + +namespace velox::daemon::rpc::platform { + +class Wakeup { +public: + Wakeup(); + ~Wakeup(); + + Wakeup(const Wakeup&) = delete; + Wakeup& operator=(const Wakeup&) = delete; + + // The fd to register with poll(2) for readability. + int pollfd() const noexcept { return fd_; } + + // Make pollfd() readable. Async-signal-safe and thread-safe. + void signal() noexcept; + + // Drain whatever signal() queued so pollfd() stops being readable. Call this from the + // loop thread once pollfd() fires. + void drain() noexcept; + +private: + int fd_ = -1; +}; + +} // namespace velox::daemon::rpc::platform diff --git a/daemon/src/rpc/runtime_dir.cpp b/daemon/src/rpc/runtime_dir.cpp index d293411..efc2a7b 100644 --- a/daemon/src/rpc/runtime_dir.cpp +++ b/daemon/src/rpc/runtime_dir.cpp @@ -5,9 +5,10 @@ #include #include -#include #include +#include "rpc/platform/runtime_paths.hpp" + namespace velox::daemon::rpc { namespace { @@ -33,17 +34,7 @@ std::error_code ensure_private_dir(const std::string& dir) { 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 (auto ec = platform::runtime_base_dir(base)) return ec; if (!base.empty() && base.back() == '/') base.pop_back(); const std::string dir = base + "/velox"; @@ -55,13 +46,7 @@ std::error_code resolve_runtime_dir(RuntimeDir& out) { std::error_code resolve_data_dir(std::string& out) { std::string base; - if (const char* xdg = ::getenv("XDG_DATA_HOME"); xdg != nullptr && xdg[0] != '\0') { - base = xdg; - } else if (const char* home = ::getenv("HOME"); home != nullptr && home[0] != '\0') { - base = std::string(home) + "/.local/share"; - } else { - return errc(ENOENT); - } + if (auto ec = platform::data_base_dir(base)) return ec; if (!base.empty() && base.back() == '/') base.pop_back(); // Create the XDG base components leniently, then the velox dir with a strict check. diff --git a/daemon/src/rpc/single_instance.hpp b/daemon/src/rpc/single_instance.hpp deleted file mode 100644 index c2b87a4..0000000 --- a/daemon/src/rpc/single_instance.hpp +++ /dev/null @@ -1,27 +0,0 @@ -#pragma once - -// Single-instance guard: bind an abstract-namespace Unix socket whose name is derived -// from the canonical runtime directory (resolve_runtime_dir's result — already the -// per-user default, /run/user//velox, unless XDG_RUNTIME_DIR says otherwise). A -// second daemon pointed at the same runtime dir gets EADDRINUSE and exits; one pointed at -// a different (isolated / test) runtime dir gets its own lock and starts fine. The kernel -// reclaims an abstract-namespace address when the holding process dies, so a crash never -// wedges it (docs/01 §2). -// -// Naming this "velox-daemon-" alone (the old scheme) meant exactly one name per -// user system-wide, so XDG_RUNTIME_DIR isolation never reached it: a leaked test veloxd -// with the same euid held the lock for every isolated instance too, real or test, until -// it was killed. Hashing the resolved runtime dir path instead keeps the real per-user -// daemon unique (its runtime dir is unique to it) while letting isolated instances that -// each point at their own runtime dir coexist. - -#include - -namespace velox::daemon::rpc { - -// Returns the held fd (kept open for the process lifetime; closing it releases the lock) -// or -1 if another process already holds the lock for this exact `runtime_dir`, or on any -// other socket error. -int acquire_single_instance_lock(const std::string& runtime_dir); - -} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp index 6b5bcb1..1048124 100644 --- a/daemon/src/rpc/uds_server.cpp +++ b/daemon/src/rpc/uds_server.cpp @@ -15,6 +15,7 @@ #include #include "rpc/event_loop.hpp" +#include "rpc/platform/peer_id.hpp" #include "rpc/systemd_activation.hpp" #include "version.hpp" @@ -140,10 +141,8 @@ void UdsServer::on_listener_readable() { 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()) { + platform::PeerId peer{}; + if (platform::peer_of(cfd, peer) || peer.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); diff --git a/daemon/tests/single_instance_test.cpp b/daemon/tests/single_instance_test.cpp index 60e5e2d..000ae29 100644 --- a/daemon/tests/single_instance_test.cpp +++ b/daemon/tests/single_instance_test.cpp @@ -5,15 +5,15 @@ #include #include "check.hpp" -#include "rpc/single_instance.hpp" +#include "rpc/platform/instance_lock.hpp" -using namespace velox::daemon::rpc; +using namespace velox::daemon::rpc::platform; void run() { // Two different runtime dirs: both acquire the lock independently. { - const int a = acquire_single_instance_lock("/run/user/1000/velox-test-a"); - const int b = acquire_single_instance_lock("/run/user/1000/velox-test-b"); + const int a = acquire_instance_lock("/run/user/1000/velox-test-a"); + const int b = acquire_instance_lock("/run/user/1000/velox-test-b"); CHECK(a >= 0); CHECK(b >= 0); if (a >= 0) ::close(a); @@ -22,16 +22,16 @@ void run() { // Same runtime dir: the second attempt is refused while the first still holds it. { - const int first = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + const int first = acquire_instance_lock("/run/user/1000/velox-test-shared"); CHECK(first >= 0); - const int second = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + const int second = acquire_instance_lock("/run/user/1000/velox-test-shared"); CHECK(second < 0); if (first >= 0) ::close(first); if (second >= 0) ::close(second); // Releasing (closing) the fd frees the abstract-namespace name immediately — a // third attempt at the same dir succeeds once the first is gone. - const int third = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + const int third = acquire_instance_lock("/run/user/1000/velox-test-shared"); CHECK(third >= 0); if (third >= 0) ::close(third); }