From d93c8e10a0ad398fa60541596d82b8ec1fe01418 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 20:28:25 +0400 Subject: [PATCH 1/2] =?UTF-8?q?daemon:=20sched/scheduler=20=E2=80=94=20gov?= =?UTF-8?q?ernor=20<->=20store=20<->=20engine,=20against=20an=20EnginePort?= =?UTF-8?q?=20seam?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Scheduler that D4 was waiting on. Built against CORE's engine HEADERS (now in main); the real EnginePort and the veloxd wiring wait for lane/core's stage-8 bodies to reach main (deferrals.md D4a/D4b) — core/src/task/ is still .gitkeep there, so linking vdm::Engine now would be an unresolved symbol. - sched/engine_port — the abstract seam: start/pause/resume/cancel/ provide_auth/decide/refresh_url + the ADR 0011 admission config (set_task_order / set_max_active_segments / set_host_segment_cap). Keeps the Scheduler testable without a live engine and the daemon unbound from the concrete vdm::Engine. - sched/fake_engine_port — a recording impl for tests. - sched/scheduler: * owns the wire-UUID <-> vdm::TaskId map. * tick(): snapshot queues (schedule window evaluated with an injectable clock) + non-terminal tasks -> governor.evaluate -> apply. to_start builds a vdm::task::DownloadSpec from the row and calls EnginePort::start; to_resume -> resume(); to_pause -> pause() + writes the pause_reason; priority_order -> set_task_order over the mapped engine ids. `new` tasks are parked (startMode manual) and skipped. * on_engine_state(wire_id, state, err): projects an engine transition onto the store row (state, pause_reason='auto' when an error rides a paused transition per ADR 0013 §2, flattened error columns) so the next tick sees ground truth. This is also the hook event.task.state will fire from (D5). * reconcile_after_restart(): CORE-owned states -> queued, paused keeps its reason (ADR 0013 §5). * reload_config(): reads connection.maxConcurrentDownloads / maxActiveSegments + a daemon-local host-cap map, pushes caps to the engine, updates the governor. * Deps: injectable local-now clock and a post_to_loop marshaller (engine callbacks arrive on engine threads; default runs inline for tests). Test veloxd.sched_scheduler (ASan+UBSan and TSan clean): admission + ordering, a slot freeing on completion, queue-stop -> pause (queue_stopped) then queue-restart -> resume (not a fresh start), engine auto-pause -> pause_reason 'auto' + never auto-resumed, reconcile_after_restart, reload_config caps push. 35 daemon/cli tests green. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig --- daemon/CMakeLists.txt | 3 +- daemon/docs/deferrals.md | 5 +- daemon/src/sched/engine_port.hpp | 46 ++++ daemon/src/sched/fake_engine_port.hpp | 55 +++++ daemon/src/sched/scheduler.cpp | 314 ++++++++++++++++++++++++++ daemon/src/sched/scheduler.hpp | 88 ++++++++ daemon/tests/CMakeLists.txt | 1 + daemon/tests/sched_scheduler_test.cpp | 191 ++++++++++++++++ 8 files changed, 700 insertions(+), 3 deletions(-) create mode 100644 daemon/src/sched/engine_port.hpp create mode 100644 daemon/src/sched/fake_engine_port.hpp create mode 100644 daemon/src/sched/scheduler.cpp create mode 100644 daemon/src/sched/scheduler.hpp create mode 100644 daemon/tests/sched_scheduler_test.cpp diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index 099c923..1a3273d 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -59,12 +59,13 @@ target_compile_options(veloxd_fs PRIVATE -Wall -Wextra -Wpedantic -Werror) add_library(veloxd_sched STATIC src/sched/schedule_window.cpp src/sched/governor.cpp + src/sched/scheduler.cpp ) add_library(velox::daemon_sched ALIAS veloxd_sched) target_include_directories(veloxd_sched PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src) target_compile_features(veloxd_sched PUBLIC cxx_std_23) target_compile_options(veloxd_sched PRIVATE -Wall -Wextra -Wpedantic -Werror) -target_link_libraries(veloxd_sched PUBLIC velox::proto nlohmann_json::nlohmann_json) +target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store nlohmann_json::nlohmann_json) # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------ add_library(veloxd_rpc STATIC diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 8bf668b..37f1ccc 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -8,5 +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 | -| D4 | `sched/` is the pure `Governor` + schedule window only; no `Scheduler` wiring to store/engine/timer | `sched/` | `Engine` bodies land in CORE stage 8; `Scheduler` needs the UUID↔`vdm::TaskId` map, a store query layer, and a timer | After CORE stage 8 lands `Engine::start()` | -| 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 | With the callback → `event.*` projection | +| 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 | +| 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/src/sched/engine_port.hpp b/daemon/src/sched/engine_port.hpp new file mode 100644 index 0000000..096fb82 --- /dev/null +++ b/daemon/src/sched/engine_port.hpp @@ -0,0 +1,46 @@ +#pragma once + +// The seam between the Scheduler and CORE's engine. Everything the Scheduler drives on the +// engine goes through this interface, so the Scheduler is unit-testable without a live +// engine and the daemon is not bound to the concrete `vdm::Engine`. The real +// implementation (engine_port_core, added when velox::core's stage-8 bodies reach main) +// wraps `vdm::Engine` + `vdm::segment::SegmentBudget`; FakeEnginePort records calls. +// +// Task ids here are `vdm::TaskId` — the engine assigns one from start() and the Scheduler +// keeps the wire-UUID <-> TaskId map (ADR 0013). Admission is the Scheduler's: it calls +// start() only for a task the governor admitted, and the engine begins probing at once +// (it does not queue). The min-1 fairness rule in SegmentBudget then guarantees each +// started task a slot; set_task_order pushes the priority. + +#include +#include +#include +#include + +#include "vdm/ids.hpp" +#include "vdm/task/download.hpp" + +namespace velox::daemon::sched { + +class EnginePort { +public: + virtual ~EnginePort() = default; + + virtual vdm::TaskId start(const vdm::task::DownloadSpec& spec, + vdm::task::DownloadCallbacks callbacks) = 0; + + virtual void pause(vdm::TaskId) = 0; + virtual void resume(vdm::TaskId) = 0; + virtual void cancel(vdm::TaskId, bool discard_partial) = 0; + virtual void provide_auth(vdm::TaskId, const std::string& username, + const std::string& password, bool remember) = 0; + virtual void decide(vdm::TaskId, vdm::task::Decision) = 0; + virtual void refresh_url(vdm::TaskId, const std::string& url) = 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; + virtual void set_host_segment_cap(const std::string& host, std::uint32_t cap) = 0; +}; + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/fake_engine_port.hpp b/daemon/src/sched/fake_engine_port.hpp new file mode 100644 index 0000000..24ec005 --- /dev/null +++ b/daemon/src/sched/fake_engine_port.hpp @@ -0,0 +1,55 @@ +#pragma once + +// A recording EnginePort for Scheduler tests. Every call is logged; start() hands back a +// sequential TaskId. No threads, no real work. + +#include +#include +#include + +#include "sched/engine_port.hpp" + +namespace velox::daemon::sched { + +class FakeEnginePort final : public EnginePort { +public: + struct StartCall { + vdm::TaskId id; + std::string url; + std::string save_path; + vdm::task::DownloadCallbacks callbacks; + }; + + std::vector starts; + std::vector paused; + std::vector resumed; + std::vector> cancelled; + std::vector> orders; + std::vector max_active_segments; + std::vector> host_caps; + + vdm::TaskId start(const vdm::task::DownloadSpec& spec, + vdm::task::DownloadCallbacks callbacks) override { + const vdm::TaskId id{next_++}; + starts.push_back({id, spec.url, spec.save_path, std::move(callbacks)}); + return id; + } + void pause(vdm::TaskId id) override { paused.push_back(id); } + void resume(vdm::TaskId id) override { resumed.push_back(id); } + void cancel(vdm::TaskId id, bool discard) override { cancelled.emplace_back(id, discard); } + 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 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 { + host_caps.emplace_back(h, c); + } + + const std::vector& last_order() const { return orders.back(); } + +private: + std::uint64_t next_ = 1; +}; + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp new file mode 100644 index 0000000..de99c37 --- /dev/null +++ b/daemon/src/sched/scheduler.cpp @@ -0,0 +1,314 @@ +#include "sched/scheduler.hpp" + +#include +#include + +#include + +#include "sched/schedule_window.hpp" +#include "store/settings.hpp" +#include "store/tasks.hpp" + +namespace velox::daemon::sched { + +namespace proto = velox::proto; + +namespace { + +std::tm local_now_default() { + const std::time_t t = std::time(nullptr); + std::tm tm{}; + ::localtime_r(&t, &tm); + return tm; +} + +// host[:port] out of a URL, lowercased. "" when it cannot be parsed (opts the task out of +// the per-host cap rather than lumping unrelated tasks under ""). +std::string host_of(std::string_view url) { + auto scheme = url.find("://"); + std::string_view rest = scheme == std::string_view::npos ? url : url.substr(scheme + 3); + const auto at = rest.find('@'); + if (at != std::string_view::npos) rest = rest.substr(at + 1); + const auto end = rest.find_first_of("/:?#"); + std::string h(end == std::string_view::npos ? rest : rest.substr(0, end)); + std::transform(h.begin(), h.end(), h.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return h; +} + +// ISO-8601 timestamp -> a monotonic integer for FIFO tiebreaking ("2026-09-10T15:57:50Z" +// -> 20260910155750). Lexical order of the digits is chronological. +std::int64_t rank_of(std::string_view iso) { + std::string digits; + for (char c : iso) + if (std::isdigit(static_cast(c))) digits.push_back(c); + digits.resize(std::min(digits.size(), 17)); // fits in int64 + return digits.empty() ? 0 : std::stoll(digits); +} + +RunState run_state_of(const std::string& s) { + if (s == "queued") return RunState::Queued; + if (s == "paused") return RunState::Paused; + if (s == "complete" || s == "failed" || s == "cancelled") return RunState::Terminal; + return RunState::Running; // probing / connecting / downloading / retry_wait / assembling / verifying +} + +std::optional pause_reason_of(const std::optional& s) { + if (!s) return std::nullopt; + if (*s == "user") return PauseReason::User; + if (*s == "schedule") return PauseReason::Schedule; + if (*s == "queue_stopped") return PauseReason::QueueStopped; + if (*s == "admission_reconcile") return PauseReason::AdmissionReconcile; + if (*s == "auto") return PauseReason::Auto; + return std::nullopt; +} + +const char* pause_reason_str(PauseReason r) { + switch (r) { + case PauseReason::User: return "user"; + case PauseReason::Schedule: return "schedule"; + case PauseReason::QueueStopped: return "queue_stopped"; + case PauseReason::AdmissionReconcile: return "admission_reconcile"; + case PauseReason::Auto: return "auto"; + } + return "user"; +} + +const char* engine_state_name(vdm::task::EngineState s) { + using E = vdm::task::EngineState; + switch (s) { + case E::probing: return "probing"; + case E::connecting: return "connecting"; + case E::downloading: return "downloading"; + case E::paused: return "paused"; + case E::retry_wait: return "retry_wait"; + case E::assembling: return "assembling"; + case E::verifying: return "verifying"; + case E::complete: return "complete"; + case E::failed: return "failed"; + case E::cancelled: return "cancelled"; + } + return "connecting"; +} + +// The non-terminal states the scheduler cares about — feeds the tasks.list filter. +std::vector non_terminal_states() { + return {proto::TaskState::New, proto::TaskState::Probing, + proto::TaskState::Queued, proto::TaskState::Connecting, + proto::TaskState::Downloading, proto::TaskState::Paused, + proto::TaskState::RetryWait, proto::TaskState::Assembling, + proto::TaskState::Verifying}; +} + +} // namespace + +Scheduler::Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps) + : db_(db), engine_(engine), governor_(std::move(governor)), deps_(std::move(deps)) { + if (!deps_.local_now) deps_.local_now = local_now_default; + if (!deps_.post_to_loop) deps_.post_to_loop = [](std::function f) { f(); }; +} + +void Scheduler::map(const std::string& wire_id, vdm::TaskId engine_id) { + to_engine_[wire_id] = engine_id; + to_wire_[engine_id] = wire_id; +} +void Scheduler::unmap_engine(vdm::TaskId engine_id) { + if (auto it = to_wire_.find(engine_id); it != to_wire_.end()) { + to_engine_.erase(it->second); + to_wire_.erase(it); + } +} +std::optional Scheduler::wire_id_of(vdm::TaskId id) const { + auto it = to_wire_.find(id); + return it == to_wire_.end() ? std::nullopt : std::optional(it->second); +} +std::optional Scheduler::engine_id_of(const std::string& wire_id) const { + auto it = to_engine_.find(wire_id); + return it == to_engine_.end() ? std::nullopt : std::optional(it->second); +} + +store::DbResult Scheduler::reconcile_after_restart() { + // Any CORE-owned state (probing..verifying) becomes `queued`; the engine knows nothing + // across a restart and will re-probe / re-resume from the .veloxpart.meta sidecar when + // the scheduler starts it again (ADR 0013 §5). `paused` and `new` are left alone. + return db_.exec( + "UPDATE tasks SET state = 'queued', pause_reason = NULL " + "WHERE state IN ('probing','connecting','downloading','retry_wait','assembling','verifying')"); +} + +store::DbResult Scheduler::reload_config() { + store::Settings settings(db_); + GovernorConfig cfg; + cfg.max_concurrent_downloads = settings.get_int("connection.maxConcurrentDownloads"); + cfg.max_active_segments = settings.get_int("connection.maxActiveSegments"); + + // Per-host caps: a JSON object {host: n} under a daemon-local key. Absent => none. + if (auto raw = settings.get_raw("saveTo.hostSegmentCaps"); raw && *raw) { + auto j = nlohmann::json::parse(**raw, nullptr, false); + if (j.is_object()) { + for (const auto& [host, n] : j.items()) { + if (n.is_number_integer()) { + const auto cap = n.get(); + cfg.host_caps[host] = cap; + engine_.set_host_segment_cap(host, static_cast(std::max(cap, 0))); + } + } + } + } + governor_.set_config(cfg); + engine_.set_max_active_segments( + static_cast(std::max(cfg.max_active_segments, 1))); + return {}; +} + +store::DbResult Scheduler::tick() { + // --- snapshot: queues ----------------------------------------------------------- + std::vector queues; + { + auto st = db_.prepare("SELECT queue_id, state, max_concurrent, schedule FROM queues"); + if (!st) return std::unexpected(st.error()); + const std::tm now = deps_.local_now(); + for (;;) { + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + QueueView q; + q.queue_id = st->column_text(0); + q.running = st->column_text(1) == "running"; + q.max_concurrent = st->column_int(2); + if (!st->column_is_null(3)) { + auto j = nlohmann::json::parse(st->column_text(3), nullptr, false); + proto::Schedule sched; + if (auto p = proto::parse(j, "schedule")) sched = *p; + q.window_open = window_open(sched, now); + } + queues.push_back(std::move(q)); + } + } + + // --- snapshot: tasks ---------------------------------------------------------- + store::Tasks tasks(db_); + proto::TaskFilter filter; + filter.states = non_terminal_states(); + auto page = tasks.list(filter, std::nullopt, 0, 100000); + if (!page) return std::unexpected(page.error()); + + std::vector views; + views.reserve(page->rows.size()); + for (const auto& r : page->rows) { + if (r.state == "new") continue; // parked until the user starts it + TaskView v; + v.task_id = r.task_id; + v.run_state = run_state_of(r.state); + v.pause_reason = pause_reason_of(r.pause_reason); + v.queue_id = r.queue_id; + v.queue_position = r.queue_position.value_or(0); + v.host = host_of(r.url); + v.admit_rank = rank_of(r.created_at); + views.push_back(std::move(v)); + } + + const Decision d = governor_.evaluate(views, queues); + + // --- apply ------------------------------------------------------------------ + for (const auto& wire_id : d.to_start) { + auto got = tasks.get(wire_id); + if (!got || !got->has_value()) continue; + const store::TaskRow& row = **got; + + vdm::task::DownloadSpec spec; + spec.url = row.url; + spec.save_path = row.save_dir + "/" + row.filename; + if (row.req_segments) spec.segments = static_cast(*row.req_segments); + if (row.req_buffer_bytes) + spec.buffer_bytes = static_cast(*row.req_buffer_bytes); + if (row.checksum_algo && row.checksum_value) { + vdm::task::Checksum ck; + ck.hex = *row.checksum_value; + if (*row.checksum_algo == "md5") ck.algo = vdm::task::Checksum::Algo::md5; + else if (*row.checksum_algo == "sha1") ck.algo = vdm::task::Checksum::Algo::sha1; + else if (*row.checksum_algo == "sha512") ck.algo = vdm::task::Checksum::Algo::sha512; + else ck.algo = vdm::task::Checksum::Algo::sha256; + spec.checksum = ck; + } + spec.allow_resume = true; // resume from a sidecar if one is beside save_path + // headers / cookies / referrer / user_agent are not persisted yet (a URL-only + // `velox add` has none); the capture path will fill them when it lands. + + vdm::task::DownloadCallbacks cbs; + const std::string id_copy = wire_id; + cbs.on_state = [this, id_copy](vdm::task::EngineState, vdm::task::EngineState to, + const std::optional& err) { + std::optional ef; + if (err) { + ef = TaskErrorFields{}; + ef->code = std::string(vdm::error_name(err->code)); // matches TaskErrorCode + ef->message = err->context; + if (err->http_status != 0) ef->http_status = err->http_status; + ef->retryable = err->retryable; + } + const std::string to_name = engine_state_name(to); + deps_.post_to_loop( + [this, id_copy, to_name, ef]() { on_engine_state(id_copy, to_name, ef); }); + }; + + const vdm::TaskId engine_id = engine_.start(spec, std::move(cbs)); + map(wire_id, engine_id); + (void)tasks.set_state(wire_id, "probing", std::nullopt); + } + + for (const auto& wire_id : d.to_resume) { + if (auto eid = engine_id_of(wire_id)) { + engine_.resume(*eid); + (void)tasks.set_state(wire_id, "connecting", std::nullopt); + } + } + + for (const auto& wire_id : d.to_pause) { + const auto reason = d.pause_reasons.count(wire_id) + ? pause_reason_str(d.pause_reasons.at(wire_id)) + : "user"; + if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid); + (void)tasks.set_state(wire_id, "paused", std::string(reason)); + } + + std::vector order; + order.reserve(d.priority_order.size()); + for (const auto& wire_id : d.priority_order) + if (auto eid = engine_id_of(wire_id)) order.push_back(*eid); + engine_.set_task_order(order); + + return {}; +} + +void Scheduler::on_engine_state(const std::string& wire_id, std::string_view engine_state, + const std::optional& err) { + store::Tasks tasks(db_); + // pause_reason: an engine-initiated pause carries an error => 'auto' (ADR 0013 §2); + // otherwise set_state clears the column. + std::optional reason; + if (engine_state == "paused" && err) reason = "auto"; + (void)tasks.set_state(wire_id, engine_state, reason); + + if (err) { + auto st = db_.prepare( + "UPDATE tasks SET error_code=?2, error_message=?3, error_http_status=?4, " + "error_retryable=?5 WHERE task_id=?1"); + if (st) { + (void)st->bind(1, std::string_view(wire_id)); + (void)st->bind(2, std::string_view(err->code)); + (void)st->bind(3, std::string_view(err->message)); + if (err->http_status) (void)st->bind(4, *err->http_status); + else (void)st->bind_null(4); + if (err->retryable) (void)st->bind(5, static_cast(*err->retryable)); + else (void)st->bind_null(5); + (void)st->step(); + } + } + + if (engine_state == "complete" || engine_state == "failed" || engine_state == "cancelled") { + if (auto eid = engine_id_of(wire_id)) unmap_engine(*eid); + } +} + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp new file mode 100644 index 0000000..0e82f9b --- /dev/null +++ b/daemon/src/sched/scheduler.hpp @@ -0,0 +1,88 @@ +#pragma once + +// The Scheduler ties the pure Governor to the store and the engine. It owns the +// wire-UUID <-> vdm::TaskId map and is the only thing that calls EnginePort::start / +// pause / resume / set_task_order. +// +// Threading: tick(), reload_config(), reconcile_after_restart() and the on_engine_* +// callbacks all run on ONE thread (the RPC loop). Engine callbacks arrive on engine +// threads, so the real wiring passes a `post_to_loop` that marshals them here; the +// default runs them inline (tests, single-threaded). +// +// Not yet wired into veloxd — that plus the real EnginePort land when velox::core's +// stage-8 bodies reach main (daemon/docs/deferrals.md D4). + +#include +#include +#include +#include +#include +#include + +#include "sched/engine_port.hpp" +#include "sched/governor.hpp" +#include "store/sqlite.hpp" +#include "vdm/ids.hpp" + +namespace velox::daemon::sched { + +// The subset of proto::TaskError the store row carries; passed by on_engine_state so a +// failed / retry_wait / auto-paused transition lands in the DB. +struct TaskErrorFields { + std::string code; // TaskErrorCode spelling + std::string message; + std::optional http_status; + std::optional retryable; + std::optional attempt; +}; + +class Scheduler { +public: + // `local_now` returns a fully-populated std::tm in local time; injected so tests can + // pin the clock. `post_to_loop` marshals an engine-thread callback onto the loop + // thread; the default calls it inline. + struct Deps { + std::function local_now; + std::function)> post_to_loop; + }; + + Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps = {}); + + // ADR 0013 §5: on daemon start, every task whose persisted state is a CORE-owned one + // (probing..verifying) is rewritten to `queued`; `paused` keeps its pauseReason. The + // engine holds no state across a restart. + store::DbResult reconcile_after_restart(); + + // Re-read connection.maxConcurrentDownloads / maxActiveSegments and the per-host cap + // table; push the caps to the engine and update the governor config. Call at startup + // and on settings.set of a connection.* key. + store::DbResult reload_config(); + + // One scheduling pass: snapshot the store, run the governor, apply its decision + // (start / resume / pause via the engine, update task state rows, push set_task_order). + store::DbResult tick(); + + // Engine lifecycle callback -> store projection, so the next tick sees ground truth. + // Keyed by wire UUID (known when the callback is built, before start() returns the + // TaskId). Also the hook the event.task.state fan-out will use — D5. + void on_engine_state(const std::string& wire_id, std::string_view engine_state, + const std::optional& err); + + // Diagnostics / tests. + std::optional wire_id_of(vdm::TaskId id) const; + std::optional engine_id_of(const std::string& wire_id) const; + +private: + void map(const std::string& wire_id, vdm::TaskId engine_id); + void unmap_engine(vdm::TaskId engine_id); + + store::Db& db_; + EnginePort& engine_; + Governor governor_; + Deps deps_; + + std::unordered_map to_engine_; + std::unordered_map to_wire_; +}; + +} // namespace velox::daemon::sched diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index 736e92e..fac594e 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -20,3 +20,4 @@ veloxd_test(sched_window LIBS veloxd_sched) veloxd_test(sched_governor LIBS veloxd_sched) veloxd_test(safepath LIBS veloxd_fs) veloxd_test(store_tasks LIBS veloxd_store) +veloxd_test(sched_scheduler LIBS veloxd_sched) diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp new file mode 100644 index 0000000..3d614b7 --- /dev/null +++ b/daemon/tests/sched_scheduler_test.cpp @@ -0,0 +1,191 @@ +// Scheduler against a FakeEnginePort and an in-memory store: admission, priority order, +// queue stop, restart reconciliation, and the engine-state -> store projection. + +#include + +#include "check.hpp" +#include "sched/fake_engine_port.hpp" +#include "sched/governor.hpp" +#include "sched/scheduler.hpp" +#include "store/migrations.hpp" +#include "store/settings.hpp" +#include "store/sqlite.hpp" +#include "store/tasks.hpp" + +using namespace velox::daemon; +using sched::FakeEnginePort; +using sched::Governor; +using sched::GovernorConfig; +using sched::Scheduler; + +namespace { + +store::TaskRow task(std::string id, std::string state, std::string created, + std::optional queue = std::nullopt, std::int64_t pos = 0) { + store::TaskRow r; + r.task_id = std::move(id); + r.url = "https://cdn.example/" + r.task_id; + r.save_dir = "/tmp"; + r.filename = r.task_id + ".bin"; + r.state = std::move(state); + r.created_at = std::move(created); + r.queue_id = std::move(queue); + if (r.queue_id) r.queue_position = pos; + return r; +} + +std::string task_state(store::Db& db, const std::string& id) { + store::Tasks t(db); + auto g = t.get(id); + return (g && *g) ? (*g)->state : std::string(""); +} + +} // namespace + +void run() { + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + store::Tasks tasks(*db); + + // --- admission: 4 queued, cap 2 -> start the 2 oldest, in order ----------------- + { + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 2, + .max_active_segments = 32})); + for (int i = 0; i < 4; ++i) + CHECK(tasks.insert(task("a" + std::to_string(i), "queued", + "2026-09-10T10:0" + std::to_string(i) + ":00Z")) + .has_value()); + + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 2u); + CHECK_EQ(engine.starts[0].url, std::string("https://cdn.example/a0")); + CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/a1")); + CHECK_EQ(engine.starts[0].save_path, std::string("/tmp/a0.bin")); + CHECK_EQ(task_state(*db, "a0"), std::string("probing")); + CHECK_EQ(task_state(*db, "a2"), std::string("queued")); // not admitted + + // set_task_order carries exactly the started tasks, oldest first. + CHECK_EQ(engine.last_order().size(), 2u); + CHECK(engine.last_order()[0] == engine.starts[0].id); + + // A second tick with no free slots starts nothing new. + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 2u); + } + + // --- engine reports downloading, then one completes -> a slot frees ------------ + { + for (const char* id : {"a0", "a1", "a2", "a3"}) tasks.remove(id); + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 1, + .max_active_segments = 32})); + CHECK(tasks.insert(task("b0", "queued", "2026-09-10T10:00:00Z")).has_value()); + CHECK(tasks.insert(task("b1", "queued", "2026-09-10T10:01:00Z")).has_value()); + + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 1u); // b0 only + CHECK_EQ(task_state(*db, "b0"), std::string("probing")); + + sched.on_engine_state("b0", "downloading", std::nullopt); + CHECK_EQ(task_state(*db, "b0"), std::string("downloading")); + + sched.on_engine_state("b0", "complete", std::nullopt); + CHECK_EQ(task_state(*db, "b0"), std::string("complete")); + + CHECK(sched.tick().has_value()); // b0 terminal -> b1 admitted + CHECK_EQ(engine.starts.size(), 2u); + CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/b1")); + } + + // --- a stopped queue: running tasks get paused with reason queue_stopped ------- + { + for (const char* id : {"b0", "b1"}) tasks.remove(id); + CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value()); + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + CHECK(tasks.insert(task("q0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value()); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 1u); + sched.on_engine_state("q0", "downloading", std::nullopt); + + CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value()); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.paused.size(), 1u); + CHECK_EQ(task_state(*db, "q0"), std::string("paused")); + store::Tasks t(*db); + CHECK_EQ(t.get("q0").value().value().pause_reason.value_or(""), std::string("queue_stopped")); + + // Restart the queue -> the task resumes (not a fresh start). + CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value()); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.resumed.size(), 1u); + CHECK_EQ(engine.starts.size(), 1u); // no new start + } + + // --- an engine auto-pause (error present) -> pause_reason 'auto', not touched -- + { + for (const char* id : {"q0"}) tasks.remove(id); + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + CHECK(tasks.insert(task("auth", "queued", "2026-09-10T10:00:00Z")).has_value()); + CHECK(sched.tick().has_value()); + + sched::TaskErrorFields ef; + ef.code = "auth_required"; + ef.message = "401"; + ef.http_status = 401; + sched.on_engine_state("auth", "paused", ef); + + store::Tasks t(*db); + auto row = t.get("auth").value().value(); + CHECK_EQ(row.state, std::string("paused")); + CHECK_EQ(row.pause_reason.value_or(""), std::string("auto")); + CHECK_EQ(row.error_code.value_or(""), std::string("auth_required")); + + // A tick must NOT resume an auto-paused task. + engine.resumed.clear(); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.resumed.size(), 0u); + } + + // --- reconcile_after_restart: CORE-owned states -> queued -------------------- + { + for (const char* id : {"auth"}) tasks.remove(id); + CHECK(tasks.insert(task("r0", "downloading", "2026-09-10T10:00:00Z")).has_value()); + CHECK(tasks.insert(task("r1", "verifying", "2026-09-10T10:01:00Z")).has_value()); + CHECK(tasks.insert(task("r2", "paused", "2026-09-10T10:02:00Z")).has_value()); + CHECK(db->exec("UPDATE tasks SET pause_reason='user' WHERE task_id='r2'").has_value()); + + FakeEnginePort engine; + Scheduler sched(*db, engine, Governor(GovernorConfig{})); + CHECK(sched.reconcile_after_restart().has_value()); + CHECK_EQ(task_state(*db, "r0"), std::string("queued")); + CHECK_EQ(task_state(*db, "r1"), std::string("queued")); + CHECK_EQ(task_state(*db, "r2"), std::string("paused")); // paused survives + store::Tasks t(*db); + CHECK_EQ(t.get("r2").value().value().pause_reason.value_or(""), std::string("user")); + } + + // --- reload_config pushes the caps to the engine --------------------------- + { + store::Settings settings(*db); + CHECK(settings.set_raw("connection.maxActiveSegments", "12").has_value()); + FakeEnginePort engine; + Scheduler sched(*db, engine, Governor(GovernorConfig{})); + CHECK(sched.reload_config().has_value()); + CHECK_EQ(engine.max_active_segments.size(), 1u); + CHECK_EQ(engine.max_active_segments.back(), 12u); + CHECK_EQ(sched.reload_config().has_value() ? 0 : 1, 0); + } +} + +TEST_MAIN() From 08d7ee92637e225efc6676b7d6784ae894edb109 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 20:36:11 +0400 Subject: [PATCH 2/2] =?UTF-8?q?daemon:=20wire=20the=20engine=20into=20velo?= =?UTF-8?q?xd=20=E2=80=94=20the=20vertical=20slice=20runs=20end=20to=20end?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CORE stage 8 merged, so vdm::Engine is linkable. This closes D4a and narrows D4b: `velox add ` now actually downloads. - sched/engine_port_core.hpp — the real EnginePort: forwards to a live vdm::Engine, keeps the DownloadHandle per task for pause/resume/ cancel/provide_auth/decide/refresh_url, drives set_task_order / set_max_active_segments / set_host_segment_cap via engine.segment_budget(). CORE confirmed the admission model: DAEMON decides when to start(); the engine's own download_task calls register_task/set_want internally — DAEMON never touches per-task budget calls. EnginePort gains release(TaskId) so the port drops a handle when the task goes terminal. - rpc/event_loop — EventLoop::post(fn): thread-safe, runs fn on the loop thread next iteration. The marshaller for engine-thread callbacks. - main.cpp — constructs vdm::Engine + EnginePortCore + Scheduler (post_to_loop = loop.post). At startup: reconcile_after_restart() (ADR 0013 §5), reload_config(), tick(). A 1 s timerfd on the loop re-runs tick() (schedule windows, missed nudges); download.add nudges via dispatcher.set_on_mutation. End-to-end verified against tools/testserver: `velox add http://127.0.0.1:.../file/512K` -> task queued -> scheduler admits -> engine downloads 524288 bytes -> complete, file on disk. First byte-path all the way through the project. safepath-adversarial.md: re-verified per its own note — CORE landed O_NOFOLLOW on the target open (core/src/io/sparse_file.cpp), so the leaf-symlink TOCTOU is now closed; residual is down to one intermediate-dir gap (documented post-M1 chase). 36 daemon/cli tests green; scheduler + uds_roundtrip TSan-clean. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig --- daemon/docs/deferrals.md | 4 +- daemon/docs/safepath-adversarial.md | 29 +++++------- daemon/src/main.cpp | 41 ++++++++++++++++ daemon/src/rpc/dispatcher.cpp | 4 +- daemon/src/rpc/dispatcher.hpp | 7 +++ daemon/src/rpc/event_loop.cpp | 19 ++++++++ daemon/src/rpc/event_loop.hpp | 10 ++++ daemon/src/sched/engine_port.hpp | 3 ++ daemon/src/sched/engine_port_core.hpp | 68 +++++++++++++++++++++++++++ daemon/src/sched/fake_engine_port.hpp | 2 + daemon/src/sched/scheduler.cpp | 5 +- 11 files changed, 171 insertions(+), 21 deletions(-) create mode 100644 daemon/src/sched/engine_port_core.hpp 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); + } } }