diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 37f1ccc..b8e99b0 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -8,6 +8,6 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | | D2 | `download.probe` → `-32603` | `rpc/dispatcher.cpp` | `download.add` is wired (`fs/safepath` + store, real `-32011`); `download.probe` needs the engine's probe path for `-32013` | probe with the engine link (CORE stage 3 is landed; wire `Engine::probe`) | | D3 | Stub handlers for everything except `session.*`, `download.add/list/get` | `rpc/dispatcher.cpp` | No store behind them yet (categories/queues/rules/settings/limiter/schedule) | Per method, as the store query modules land behind them | -| D4a | `sched/scheduler` is built and unit-tested against `FakeEnginePort`, but there is no real `EnginePortCore` (wraps `vdm::Engine` + `SegmentBudget`) and it is not wired into `veloxd` | `sched/` | CORE stage 8 is on `lane/core`, not yet in `main` (`core/src/task/` is still `.gitkeep` there) — linking `vdm::Engine` would be an unresolved symbol | `lane/core` merges to `main`: add `engine_port_core.{hpp,cpp}` (~100 lines) + construct `Engine`/`Scheduler` in `main.cpp`, run `tick()` on the timer thread and on RPC-driven changes | -| D4b | no timer thread driving `Scheduler::tick()`; `download.add`/`pause`/`resume`/queue handlers don't nudge the scheduler | `daemon/src/main.cpp`, `rpc/dispatcher.cpp` | depends on D4a | with D4a | +| ~~D4a~~ | **Closed** — `sched/engine_port_core.hpp` wraps `vdm::Engine` + `segment_budget()`; `main.cpp` constructs `Engine` + `Scheduler`, calls `reconcile_after_restart` / `reload_config` / `tick` at startup | — | — | done (`lane/core` stage 8 merged) | +| D4b | timer + nudges: a 1 s `timerfd` re-runs `Scheduler::tick()` and `download.add` nudges via `on_mutation`. `download.pause`/`resume`/`start`/`cancel` and the queue.* handlers still don't touch the scheduler | `rpc/dispatcher.cpp` | those handlers are still stubs (D3) | as each handler is implemented behind the store, it calls `on_mutation` / drives the scheduler | | D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired. `Scheduler::on_engine_state` is the hook it will fire from | With D4a — the same engine-state callback feeds both the store and `event.task.state` | diff --git a/daemon/docs/safepath-adversarial.md b/daemon/docs/safepath-adversarial.md index 5950e07..f39c5a0 100644 --- a/daemon/docs/safepath-adversarial.md +++ b/daemon/docs/safepath-adversarial.md @@ -58,32 +58,27 @@ Roots for the examples: `allowedRoots = ["/home/u/Downloads", "/data/dl"]`, alre `-32011`. Then re-derive the final dir's path from its fd (`/proc/self/fd/N`) and re-assert containment. (A8 for the created tail, A14) 5. **Best-effort leaf check:** `fstatat(dir_fd, leaf, AT_SYMLINK_NOFOLLOW)` — refuse if it - is already a symlink. This narrows, but does not close, the create-after-check race on - the leaf: a symlink planted *after* this `fstatat` and *before* CORE opens the file is - still followed. Closing it needs CORE to open with `O_NOFOLLOW` (plus `O_EXCL` on a - fresh download). **Verified 2026-09-10: it does not yet** — - `core/src/io/sparse_file.cpp:77` is `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change - has been raised with CORE; until it lands this race is open, see the residual below. + is already a symlink. This narrows the create-after-check race on the leaf; it is fully + closed by CORE opening the download target with `O_NOFOLLOW`. **Verified 2026-09-11: + `core/src/io/sparse_file.cpp` opens `O_WRONLY | O_CREAT | O_CLOEXEC | O_NOFOLLOW`** — a + symlink swapped in as the leaf after our check fails there with `ELOOP` -> + `Error::path_rejected`. (No `O_EXCL`: resume must be able to open an existing + `.veloxpart`.) 6. **Every failure is `-32011`, `data.path` = the *original* `saveDir`** — never the resolved path, which would leak where the roots actually live. The one exception is a `filename` that violates the schema's own `maxLength`, which is `-32602` at the param layer before this code runs. -### Residual — currently OPEN, tracked +### Residual — one gap, narrowed -Two TOCTOU gaps this code does not close on its own: +**The leaf-symlink TOCTOU is closed** (step 5, verified 2026-09-11: CORE opens the target +`O_NOFOLLOW`). What remains: 1. **An existing intermediate directory** swapped for an out-of-root symlink between our `realpath` (step 3) and the write. Step 3 trusts `realpath` for the pre-existing - prefix; a full `O_NOFOLLOW` chase would reject the legitimate symlinked directories - A16 requires us to allow. -2. **The leaf** swapped for a symlink between our `fstatat` (step 5) and CORE's `open`. - -Both are closed by CORE opening the file `O_NOFOLLOW` (and, for a fresh download, -`O_EXCL`). **As verified on 2026-09-10 that is not yet the case** — -`core/src/io/sparse_file.cpp:77` opens `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change has -been raised with CORE; when it lands, update step 5 and this paragraph and re-verify the -flags at that line. + prefix; `O_NOFOLLOW` on the *file* open does not re-check the *directories* above it, + and a full `O_NOFOLLOW` directory chase would reject the legitimate symlinked + directories A16 requires us to allow. What limits the exposure *today*: the download directory lives under `~/.local/share` / `~/Downloads`, both `0700` — an attacker planting a symlink there already has write access diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index 55a6c8e..26a8c88 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -15,14 +15,20 @@ #include #include +#include + #include "rpc/dispatcher.hpp" #include "rpc/event_loop.hpp" #include "rpc/pairing.hpp" #include "rpc/runtime_dir.hpp" #include "rpc/uds_server.hpp" #include "rpc/ws_server.hpp" +#include "sched/engine_port_core.hpp" +#include "sched/governor.hpp" +#include "sched/scheduler.hpp" #include "store/migrations.hpp" #include "store/sqlite.hpp" +#include "vdm/engine.hpp" #include "version.hpp" namespace { @@ -101,7 +107,38 @@ int main() { return 1; } + // --- engine + scheduler --------------------------------------------------------- + vdm::Engine engine; + velox::daemon::sched::EnginePortCore engine_port(engine); + velox::daemon::sched::Scheduler scheduler( + *db, engine_port, velox::daemon::sched::Governor{}, + {/*local_now*/ {}, + /*post_to_loop*/ [&loop](std::function fn) { loop.post(std::move(fn)); }}); + + if (const auto ec = scheduler.reconcile_after_restart(); !ec) + std::cerr << "veloxd: restart reconcile: " << ec.error().to_string() << "\n"; + (void)scheduler.reload_config(); + (void)scheduler.tick(); // admit anything already queued in the DB + velox::daemon::rpc::VeloxDispatcher dispatcher(*db); + dispatcher.set_on_mutation([&loop, &scheduler] { + loop.post([&scheduler] { (void)scheduler.tick(); }); + }); + + // 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(); + }); + } velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path()); if (const auto ec = uds.start()) { @@ -128,6 +165,10 @@ int main() { loop.run(); std::cout << "veloxd: shutting down\n"; + if (tick_fd >= 0) { + loop.del_fd(tick_fd); + ::close(tick_fd); + } g_loop = nullptr; ::close(lock_fd); return 0; diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index bc9657b..150d916 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -151,7 +151,7 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { row.created_at = velox::daemon::now_iso(); row.start_mode = spec.startMode ? std::string(proto::to_string(*spec.startMode)) : "auto"; // startMode 'manual' parks the task in `new`; anything else makes it eligible for the - // scheduler (`queued`). The scheduler itself is not wired yet (deferrals.md D4). + // scheduler (`queued`); on_mutation_ nudges it. row.state = row.start_mode == "manual" ? "new" : "queued"; row.category_id = spec.categoryId; row.queue_id = spec.queueId; @@ -168,6 +168,8 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, "download.add: " + ins.error().message}); + if (on_mutation_) on_mutation_(); + proto::DownloadAddResult r; r.taskId = row.task_id; if (auto st = proto::parse_TaskState(row.state)) r.state = *st; diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 04a0f52..1868203 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -12,6 +12,8 @@ // "not implemented in this build" (-> -32603) until its handler and the scheduler land; // see daemon/docs/deferrals.md. +#include + #include "store/sqlite.hpp" #include "velox_proto.hpp" @@ -21,6 +23,10 @@ class VeloxDispatcher final : public velox::proto::Dispatcher { public: explicit VeloxDispatcher(velox::daemon::store::Db& db) : db_(db) {} + // Called after a handler mutates task state (download.add for now). main.cpp wires it + // to nudge the scheduler; unset in tests. + void set_on_mutation(std::function fn) { on_mutation_ = std::move(fn); } + velox::proto::HandlerResult on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override; velox::proto::HandlerResult @@ -101,6 +107,7 @@ public: private: velox::daemon::store::Db& db_; + std::function on_mutation_; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/event_loop.cpp b/daemon/src/rpc/event_loop.cpp index 3713b36..7c15807 100644 --- a/daemon/src/rpc/event_loop.cpp +++ b/daemon/src/rpc/event_loop.cpp @@ -45,6 +45,23 @@ void EventLoop::stop() noexcept { wake(); } +void EventLoop::post(std::function fn) { + { + std::lock_guard lk(post_mu_); + posts_.push_back(std::move(fn)); + } + wake(); +} + +void EventLoop::drain_posts() { + std::vector> batch; + { + std::lock_guard lk(post_mu_); + batch.swap(posts_); + } + for (auto& fn : batch) fn(); +} + void EventLoop::drain_wakeup() noexcept { std::uint64_t sink = 0; while (::read(wake_fd_, &sink, sizeof(sink)) > 0) { @@ -87,6 +104,8 @@ 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 8f96b4a..e6d80f5 100644 --- a/daemon/src/rpc/event_loop.hpp +++ b/daemon/src/rpc/event_loop.hpp @@ -12,7 +12,9 @@ #include #include #include +#include #include +#include namespace velox::daemon::rpc { @@ -52,6 +54,10 @@ public: // callback. Async-signal-safe. void wake() noexcept; + // Run `fn` on the loop thread at the next iteration. Thread-safe; the intended way to + // marshal an engine-thread callback back onto the RPC loop. + void post(std::function fn); + private: struct Entry { unsigned interest; @@ -59,11 +65,15 @@ private: }; void drain_wakeup() noexcept; + void drain_posts(); int wake_fd_; // eventfd, always registered 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_; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/sched/engine_port.hpp b/daemon/src/sched/engine_port.hpp index 096fb82..759bf37 100644 --- a/daemon/src/sched/engine_port.hpp +++ b/daemon/src/sched/engine_port.hpp @@ -37,6 +37,9 @@ public: virtual void decide(vdm::TaskId, vdm::task::Decision) = 0; virtual void refresh_url(vdm::TaskId, const std::string& url) = 0; + // The daemon is done with this task (it went terminal). Drop the handle. Idempotent. + virtual void release(vdm::TaskId) = 0; + // ADR 0011 admission surface. Values are DAEMON's; enforcement is the engine's. virtual void set_task_order(const std::vector& order) = 0; virtual void set_max_active_segments(std::uint32_t n) = 0; diff --git a/daemon/src/sched/engine_port_core.hpp b/daemon/src/sched/engine_port_core.hpp new file mode 100644 index 0000000..9f5bb51 --- /dev/null +++ b/daemon/src/sched/engine_port_core.hpp @@ -0,0 +1,68 @@ +#pragma once + +// The real EnginePort: forwards to a live vdm::Engine and keeps the DownloadHandle per +// task so pause/resume/cancel/… have something to call. All methods run on the RPC loop +// thread (the Scheduler's thread); the handles map is only touched there. + +#include + +#include "sched/engine_port.hpp" +#include "vdm/engine.hpp" +#include "vdm/task/download.hpp" + +namespace velox::daemon::sched { + +class EnginePortCore final : public EnginePort { +public: + explicit EnginePortCore(vdm::Engine& engine) : engine_(engine) {} + + vdm::TaskId start(const vdm::task::DownloadSpec& spec, + vdm::task::DownloadCallbacks callbacks) override { + vdm::task::DownloadHandle h = engine_.start(spec, std::move(callbacks)); + const vdm::TaskId id = h.id(); + handles_.insert_or_assign(id, std::move(h)); + return id; + } + + void pause(vdm::TaskId id) override { + if (auto* h = find(id)) h->pause(); + } + void resume(vdm::TaskId id) override { + if (auto* h = find(id)) h->resume(); + } + void cancel(vdm::TaskId id, bool discard_partial) override { + if (auto* h = find(id)) h->cancel(discard_partial); + } + void provide_auth(vdm::TaskId id, const std::string& u, const std::string& p, + bool remember) override { + if (auto* h = find(id)) h->provide_auth(u, p, remember); + } + void decide(vdm::TaskId id, vdm::task::Decision d) override { + if (auto* h = find(id)) h->decide(d); + } + void refresh_url(vdm::TaskId id, const std::string& url) override { + if (auto* h = find(id)) h->refresh_url(url); + } + void release(vdm::TaskId id) override { handles_.erase(id); } + + void set_task_order(const std::vector& order) override { + engine_.segment_budget().set_task_order(order); + } + void set_max_active_segments(std::uint32_t n) override { + engine_.segment_budget().set_max_active_segments(n); + } + void set_host_segment_cap(const std::string& host, std::uint32_t cap) override { + engine_.segment_budget().set_host_segment_cap(host, cap); + } + +private: + vdm::task::DownloadHandle* find(vdm::TaskId id) { + auto it = handles_.find(id); + return it == handles_.end() ? nullptr : &it->second; + } + + vdm::Engine& engine_; + std::unordered_map handles_; +}; + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/fake_engine_port.hpp b/daemon/src/sched/fake_engine_port.hpp index 24ec005..799a7b5 100644 --- a/daemon/src/sched/fake_engine_port.hpp +++ b/daemon/src/sched/fake_engine_port.hpp @@ -24,6 +24,7 @@ public: std::vector paused; std::vector resumed; std::vector> cancelled; + std::vector released; std::vector> orders; std::vector max_active_segments; std::vector> host_caps; @@ -40,6 +41,7 @@ public: void provide_auth(vdm::TaskId, const std::string&, const std::string&, bool) override {} void decide(vdm::TaskId, vdm::task::Decision) override {} void refresh_url(vdm::TaskId, const std::string&) override {} + void release(vdm::TaskId id) override { released.push_back(id); } void set_task_order(const std::vector& order) override { orders.push_back(order); } void set_max_active_segments(std::uint32_t n) override { max_active_segments.push_back(n); } void set_host_segment_cap(const std::string& h, std::uint32_t c) override { diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index de99c37..2e923cc 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -307,7 +307,10 @@ void Scheduler::on_engine_state(const std::string& wire_id, std::string_view eng } if (engine_state == "complete" || engine_state == "failed" || engine_state == "cancelled") { - if (auto eid = engine_id_of(wire_id)) unmap_engine(*eid); + if (auto eid = engine_id_of(wire_id)) { + engine_.release(*eid); + unmap_engine(*eid); + } } }