daemon: Phase 0 platform seams (ADR 0020) — eventfd/SO_PEERCRED/instance-lock/XDG behind platform/, zero behaviour change
Introduces the four rpc-layer seams docs/08-porting.md calls for this lane: platform::Wakeup (eventfd), platform::peer_of (SO_PEERCRED/struct ucred, same-UID check preserved unchanged), platform::acquire_instance_lock (abstract-namespace socket), platform::runtime_base_dir/data_base_dir (XDG lookups). Today's Linux code moves unchanged into daemon/src/rpc/platform/linux/; the seam headers carry no OS types and no #ifdef. No fifth interface for timerfd: EventLoop gains a portable add_timer() that folds the next deadline into poll()'s own timeout, replacing both timerfd instances in main.cpp — the loop already computes a deadline, so this needs no per-OS backend at all. Full suite 59/59 green; no #ifdef outside platform/linux/, no behaviour change. Co-Authored-By: Claude Sonnet 5 <[email protected]>
This commit is contained in:
+30
-57
@@ -5,6 +5,7 @@
|
||||
// (AGENT-DAEMON.md build order, steps 1 and 3). The scheduler and the engine link land
|
||||
// next.
|
||||
|
||||
#include <chrono>
|
||||
#include <csignal>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
@@ -15,16 +16,14 @@
|
||||
#include <sys/un.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <sys/timerfd.h>
|
||||
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
#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;
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
#include "rpc/event_loop.hpp"
|
||||
|
||||
#include <poll.h>
|
||||
#include <sys/eventfd.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <cerrno>
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
@@ -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<void()> 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<std::chrono::milliseconds>(soonest - now);
|
||||
return ms.count() > 0 ? static_cast<int>(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<TimerId> 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
|
||||
|
||||
@@ -10,12 +10,15 @@
|
||||
// from any thread or a signal handler — they only write() a byte to an internal eventfd.
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <functional>
|
||||
#include <mutex>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
#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<void()> 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<void()> 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<void()> 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<bool> stop_requested_ = false; // set from stop(), read by run()
|
||||
std::unordered_map<int, Entry> fds_;
|
||||
|
||||
std::mutex post_mu_;
|
||||
std::vector<std::function<void()>> posts_;
|
||||
|
||||
TimerId next_timer_id_ = 1;
|
||||
std::unordered_map<TimerId, Timer> timers_;
|
||||
};
|
||||
|
||||
} // namespace velox::daemon::rpc
|
||||
|
||||
@@ -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 <string>
|
||||
#include <system_error>
|
||||
|
||||
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
|
||||
+4
-4
@@ -1,4 +1,4 @@
|
||||
#include "rpc/single_instance.hpp"
|
||||
#include "rpc/platform/instance_lock.hpp"
|
||||
|
||||
#include <cstddef>
|
||||
#include <cstring>
|
||||
@@ -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
|
||||
@@ -0,0 +1,17 @@
|
||||
#include "rpc/platform/peer_id.hpp"
|
||||
|
||||
#include <sys/socket.h>
|
||||
|
||||
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
|
||||
@@ -0,0 +1,44 @@
|
||||
#include "rpc/platform/runtime_paths.hpp"
|
||||
|
||||
#include <sys/stat.h>
|
||||
#include <sys/types.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cerrno>
|
||||
#include <cstdlib>
|
||||
|
||||
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/<uid>: 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
|
||||
@@ -0,0 +1,32 @@
|
||||
#include "rpc/platform/wakeup.hpp"
|
||||
|
||||
#include <sys/eventfd.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
|
||||
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
|
||||
@@ -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 <system_error>
|
||||
|
||||
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
|
||||
@@ -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 <string>
|
||||
#include <system_error>
|
||||
|
||||
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/<uid>" 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
|
||||
@@ -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/<os>/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
|
||||
@@ -5,9 +5,10 @@
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cerrno>
|
||||
#include <cstdlib>
|
||||
#include <string>
|
||||
|
||||
#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/<uid>: 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.
|
||||
|
||||
@@ -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/<uid>/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-<euid>" 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 <string>
|
||||
|
||||
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
|
||||
@@ -15,6 +15,7 @@
|
||||
#include <fcntl.h>
|
||||
|
||||
#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);
|
||||
|
||||
Reference in New Issue
Block a user