From b01013942106f3acee266555de4f0e2dfe7188cf Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 19:06:19 +0400 Subject: [PATCH] =?UTF-8?q?daemon:=20sched/=20=E2=80=94=20the=20concurrenc?= =?UTF-8?q?y=20governor=20+=20schedule-window=20evaluation=20(build=20step?= =?UTF-8?q?=204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scheduling brain, built against CORE's headers (vdm/engine.hpp, vdm/segment/budget.hpp) — signatures only; the Engine bodies land in CORE stage 8 and the Scheduler that wires governor <-> store <-> engine <-> timer comes after that (daemon/docs/deferrals.md D4). - sched/governor — a pure decision function. In: a snapshot of every task's coarse RunState and every queue's state (schedule windows pre-resolved). Out: {to_start, to_resume, to_pause, pause_reasons, priority_order}. Enforces, all in TASK units per ADR 0011 §1: connection.maxConcurrentDownloads; the min(that, maxActiveSegments) clamp (§2); Queue.maxConcurrent; the per-host task cap (§4); and a stopped queue / closed window runs nothing. Never touches a task paused for `user` or CORE's `auto` (ADR 0013 §3) — only Schedule / QueueStopped / AdmissionReconcile are auto-resumable. Deterministic: main-list before queued-in-queue, then queue order, then FIFO, then task_id. - sched/schedule_window — window_open(Schedule, local tm): disabled => always open; `once` => date + time match; `periodic` => weekday in daysOfWeek (empty = every day) + time in [start, stop); null start => midnight, null stop => end of day, stop < start => overnight window. Pure; re-evaluated every tick, no cached instants. - veloxd_sched static lib; veloxd links it (nothing calls it yet). Tests (ASan+UBSan and TSan clean): veloxd.sched_window (10 window cases incl. overnight, once, null bounds), veloxd.sched_governor (global/clamp/per-queue/per-host caps, stop vs window pause reasons, resume-only-governor-reasons, auth-pause untouched, admission reconcile, determinism under shuffled input). 32 daemon/cli tests green; full tree green. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig --- daemon/CMakeLists.txt | 14 ++- daemon/docs/deferrals.md | 12 ++ daemon/src/sched/.gitkeep | 0 daemon/src/sched/governor.cpp | 158 ++++++++++++++++++++++++ daemon/src/sched/governor.hpp | 89 ++++++++++++++ daemon/src/sched/schedule_window.cpp | 89 ++++++++++++++ daemon/src/sched/schedule_window.hpp | 33 +++++ daemon/tests/CMakeLists.txt | 2 + daemon/tests/sched_governor_test.cpp | 178 +++++++++++++++++++++++++++ daemon/tests/sched_window_test.cpp | 120 ++++++++++++++++++ 10 files changed, 694 insertions(+), 1 deletion(-) create mode 100644 daemon/docs/deferrals.md delete mode 100644 daemon/src/sched/.gitkeep create mode 100644 daemon/src/sched/governor.cpp create mode 100644 daemon/src/sched/governor.hpp create mode 100644 daemon/src/sched/schedule_window.cpp create mode 100644 daemon/src/sched/schedule_window.hpp create mode 100644 daemon/tests/sched_governor_test.cpp create mode 100644 daemon/tests/sched_window_test.cpp diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index cae6335..5121f0e 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -44,6 +44,18 @@ target_compile_features(veloxd_store PUBLIC cxx_std_23) target_compile_options(veloxd_store PRIVATE -Wall -Wextra -Wpedantic -Werror) target_link_libraries(veloxd_store PUBLIC SQLite::SQLite3 PRIVATE OpenSSL::Crypto) +# --- veloxd_sched — the concurrency governor (pure; no engine link yet, see +# daemon/docs/deferrals.md D4) --------------------------------------------------- +add_library(veloxd_sched STATIC + src/sched/schedule_window.cpp + src/sched/governor.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) + # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------ add_library(veloxd_rpc STATIC src/rpc/runtime_dir.cpp @@ -68,7 +80,7 @@ target_link_libraries(veloxd_rpc add_executable(veloxd src/main.cpp) target_compile_features(veloxd PRIVATE cxx_std_23) target_compile_options(veloxd PRIVATE -Wall -Wextra -Wpedantic -Werror) -target_link_libraries(veloxd PRIVATE veloxd_rpc) +target_link_libraries(veloxd PRIVATE veloxd_rpc veloxd_sched) if(VELOX_BUILD_TESTS AND EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/CMakeLists.txt) add_subdirectory(tests) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md new file mode 100644 index 0000000..6529b76 --- /dev/null +++ b/daemon/docs/deferrals.md @@ -0,0 +1,12 @@ +# DAEMON — deferred work, tracked + +Things that are deliberately incomplete in `daemon/` right now, with why and when they +close. Kept here (not buried in commit messages) so the next pass can see them at a glance. + +| # | What | Where | Why deferred | Closes when | +|---|---|---|---|---| +| 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.add` → `-32603`, `download.probe` → `-32603` | `rpc/dispatcher.cpp` | Need path canonicalisation + allowed-root check (`-32011`) and the probe path (`-32013`); those need the engine link | `download.add` glue (after `sched/`) | +| D3 | Stub handlers for everything except `session.*`, `download.list`, `download.get` | `rpc/dispatcher.cpp` | No store behind them yet | Per method, as the store/scheduler wire in | +| 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 | diff --git a/daemon/src/sched/.gitkeep b/daemon/src/sched/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/daemon/src/sched/governor.cpp b/daemon/src/sched/governor.cpp new file mode 100644 index 0000000..b34cd04 --- /dev/null +++ b/daemon/src/sched/governor.cpp @@ -0,0 +1,158 @@ +#include "sched/governor.hpp" + +#include +#include +#include +#include + +namespace velox::daemon::sched { + +namespace { + +constexpr std::int64_t kUnlimited = std::numeric_limits::max(); + +// A total order over tasks: main-list (no queue) outranks queued-in-queue; within a queue +// by run order; then FIFO by admission rank; task_id breaks any remaining tie so the +// result is deterministic. "Lower" == higher priority == admitted first, paused last. +auto priority_key(const TaskView& t) { + const int tier = t.queue_id ? 1 : 0; + const std::string& q = t.queue_id ? *t.queue_id : ""; + return std::make_tuple(tier, q, t.queue_position, t.admit_rank, t.task_id); +} + +bool less_priority(const TaskView* a, const TaskView* b) { + return priority_key(*a) < priority_key(*b); +} + +} // namespace + +Decision Governor::evaluate(const std::vector& tasks, + const std::vector& queues) const { + Decision d; + + std::unordered_map qby; + for (const auto& q : queues) qby.emplace(q.queue_id, &q); + + const std::int64_t global_cap = + std::min(cfg_.max_concurrent_downloads, cfg_.max_active_segments); + + auto queue_runnable = [&](const std::optional& qid) -> bool { + if (!qid) return true; // the main list is always "runnable" + const auto it = qby.find(*qid); + return it != qby.end() && it->second->running && it->second->window_open; + }; + auto queue_cap = [&](const std::string& qid) -> std::int64_t { + const auto it = qby.find(qid); + return it == qby.end() ? 0 : std::max(1, it->second->max_concurrent); + }; + auto host_cap = [&](const std::string& host) -> std::int64_t { + if (host.empty()) return kUnlimited; + const auto it = cfg_.host_caps.find(host); + return it == cfg_.host_caps.end() ? kUnlimited : it->second; + }; + + // ---- phase 1: pause running tasks whose queue can no longer host them ------------- + std::vector running; + for (const auto& t : tasks) { + if (t.run_state == RunState::Running) running.push_back(&t); + } + std::sort(running.begin(), running.end(), less_priority); + + std::vector survivors; + for (const auto* t : running) { + if (t->queue_id && !queue_runnable(t->queue_id)) { + const auto it = qby.find(*t->queue_id); + const bool stopped_not_windowed = + it != qby.end() && it->second->running && !it->second->window_open; + const PauseReason why = + stopped_not_windowed ? PauseReason::Schedule : PauseReason::QueueStopped; + d.to_pause.push_back(t->task_id); + d.pause_reasons.emplace(t->task_id, why); + } else { + survivors.push_back(t); + } + } + + // ---- phase 2a: per-queue reconcile (a lowered Queue.maxConcurrent) --------------- + { + std::unordered_map per_queue; + std::vector kept; + for (const auto* t : survivors) { // already priority-sorted + if (!t->queue_id) { + kept.push_back(t); + continue; + } + auto& n = per_queue[*t->queue_id]; + if (n < queue_cap(*t->queue_id)) { + ++n; + kept.push_back(t); + } else { + d.to_pause.push_back(t->task_id); + d.pause_reasons.emplace(t->task_id, PauseReason::AdmissionReconcile); + } + } + survivors.swap(kept); + } + + // ---- phase 2b: global reconcile (a lowered maxConcurrentDownloads / clamp) ------- + if (static_cast(survivors.size()) > global_cap) { + for (std::size_t i = static_cast(std::max(global_cap, 0)); + i < survivors.size(); ++i) { + d.to_pause.push_back(survivors[i]->task_id); + d.pause_reasons.emplace(survivors[i]->task_id, PauseReason::AdmissionReconcile); + } + survivors.resize(static_cast(std::max(global_cap, 0))); + } + + // ---- phase 3: fill free slots from Queued + governor-owned Paused ---------------- + std::int64_t global_running = static_cast(survivors.size()); + std::unordered_map per_queue_running; + std::unordered_map per_host_running; + for (const auto* t : survivors) { + if (t->queue_id) ++per_queue_running[*t->queue_id]; + if (!t->host.empty()) ++per_host_running[t->host]; + } + + auto resumable = [](const TaskView& t) { + return t.run_state == RunState::Paused && t.pause_reason && + (*t.pause_reason == PauseReason::Schedule || + *t.pause_reason == PauseReason::QueueStopped || + *t.pause_reason == PauseReason::AdmissionReconcile); + }; + + std::vector candidates; + for (const auto& t : tasks) { + if (t.run_state == RunState::Queued || resumable(t)) { + if (queue_runnable(t.queue_id)) candidates.push_back(&t); + } + } + std::sort(candidates.begin(), candidates.end(), less_priority); + + std::vector newly_running; + for (const auto* c : candidates) { + if (global_running >= global_cap) break; + if (c->queue_id && per_queue_running[*c->queue_id] >= queue_cap(*c->queue_id)) continue; + if (!c->host.empty() && per_host_running[c->host] >= host_cap(c->host)) continue; + + if (c->run_state == RunState::Queued) { + d.to_start.push_back(c->task_id); + } else { + d.to_resume.push_back(c->task_id); + } + newly_running.push_back(c); + ++global_running; + if (c->queue_id) ++per_queue_running[*c->queue_id]; + if (!c->host.empty()) ++per_host_running[c->host]; + } + + // ---- priority order: everyone who will be running after this decision ------------ + std::vector will_run = survivors; + will_run.insert(will_run.end(), newly_running.begin(), newly_running.end()); + std::sort(will_run.begin(), will_run.end(), less_priority); + d.priority_order.reserve(will_run.size()); + for (const auto* t : will_run) d.priority_order.push_back(t->task_id); + + return d; +} + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/governor.hpp b/daemon/src/sched/governor.hpp new file mode 100644 index 0000000..04fd328 --- /dev/null +++ b/daemon/src/sched/governor.hpp @@ -0,0 +1,89 @@ +#pragma once + +// The concurrency governor — a pure decision function. Given a snapshot of every task's +// coarse run-state and every queue's state, it decides which tasks should start, which +// should resume, which should pause, and the priority order to push to CORE's segment +// budget. No I/O, no store, no engine, no clock — the caller supplies an already-evaluated +// world (schedule windows resolved via sched/schedule_window.hpp) and applies the result. +// +// Axes it enforces, all in TASK units (ADR 0011 §1: DAEMON counts tasks, CORE counts +// segments): +// * connection.maxConcurrentDownloads — global running-task ceiling +// * min(that, connection.maxActiveSegments) — the one ADR 0011 §2 clamp +// * Queue.maxConcurrent — per-queue running-task ceiling +// * per-host task cap (== that host's segment cap) — ADR 0011 §4 +// * queue running state + schedule window — a stopped/closed queue runs nothing +// +// What it never does: touch a task paused for a reason it does not own (user, or CORE's +// auto-pause for auth_required / server_file_changed / disk_full) — ADR 0013 §3. + +#include +#include +#include +#include +#include + +namespace velox::daemon::sched { + +// The governor's coarse view of a task. Derived from the SQLite row + the engine's last +// reported state; the governor does not need the fine-grained EngineState. +enum class RunState { + Queued, // wants a slot; eligible for admission + Running, // probing / connecting / downloading / retry_wait / assembling / verifying + Paused, // see pause_reason for whether the governor may resume it + Terminal, // complete / failed / cancelled — ignored +}; + +// Why a paused task is paused. Only the first three are governor-owned and auto-resumable; +// `User` and `Auto` (CORE's auth/decision/disk auto-pause) are never touched here. +enum class PauseReason { User, Schedule, QueueStopped, AdmissionReconcile, Auto }; + +struct TaskView { + std::string task_id; // wire UUID + RunState run_state = RunState::Queued; + std::optional pause_reason; // set iff run_state == Paused + std::optional queue_id; // nullopt => the main list (no queue) + std::int64_t queue_position = 0; // order within the queue + std::string host; // for the per-host cap; "" opts out + std::int64_t admit_rank = 0; // FIFO tiebreak (created_at, then rowid) +}; + +struct QueueView { + std::string queue_id; + bool running = false; // Queue.state == "running" + std::int64_t max_concurrent = 1; + bool window_open = true; // sched/schedule_window.hpp result for right now +}; + +struct GovernorConfig { + std::int64_t max_concurrent_downloads = 5; // connection.maxConcurrentDownloads + std::int64_t max_active_segments = 32; // connection.maxActiveSegments (§2 clamp) + std::map host_caps{}; // host -> max running tasks; absent => unlimited +}; + +struct Decision { + std::vector to_start; // Queued -> admit: caller invokes Engine::start() + std::vector to_resume; // Paused (governor-owned reason) -> Engine::resume() + std::vector to_pause; // Running -> Engine::pause(); reason in pause_reasons + std::map pause_reasons; // task_id -> why, for the DB column + std::vector priority_order; // running + starting + resuming, for set_task_order() +}; + +class Governor { +public: + Governor() = default; + explicit Governor(GovernorConfig cfg) : cfg_(std::move(cfg)) {} + + void set_config(GovernorConfig cfg) { cfg_ = std::move(cfg); } + const GovernorConfig& config() const noexcept { return cfg_; } + + // Pure: same inputs -> same Decision. `tasks` and `queues` are snapshots; order within + // them does not matter (the governor sorts by its own keys). + Decision evaluate(const std::vector& tasks, + const std::vector& queues) const; + +private: + GovernorConfig cfg_; +}; + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/schedule_window.cpp b/daemon/src/sched/schedule_window.cpp new file mode 100644 index 0000000..280f260 --- /dev/null +++ b/daemon/src/sched/schedule_window.cpp @@ -0,0 +1,89 @@ +#include "sched/schedule_window.hpp" + +#include +#include +#include +#include + +namespace velox::daemon::sched { + +namespace proto = velox::proto; + +namespace { + +// "HH:MM" -> minutes since midnight. Returns nullopt on a malformed value (the schema +// pattern should prevent that, but the schedule can also come from an older DB row). +std::optional minutes_of(std::string_view hhmm) { + if (hhmm.size() != 5 || hhmm[2] != ':') return std::nullopt; + int h = 0; + int m = 0; + if (std::from_chars(hhmm.data(), hhmm.data() + 2, h).ec != std::errc{}) return std::nullopt; + if (std::from_chars(hhmm.data() + 3, hhmm.data() + 5, m).ec != std::errc{}) return std::nullopt; + if (h < 0 || h > 23 || m < 0 || m > 59) return std::nullopt; + return h * 60 + m; +} + +bool in_window(int now_min, std::optional start_min, std::optional stop_min) { + const int start = start_min.value_or(0); + if (!stop_min) return now_min >= start; // null stop => until end of day + const int stop = *stop_min; + if (start <= stop) return now_min >= start && now_min < stop; + return now_min >= start || now_min < stop; // overnight window +} + +// tm_year/mon/mday -> "YYYY-MM-DD" to compare against onceDate. +std::array iso_date(const std::tm& t) { + std::array out{}; + const int y = t.tm_year + 1900; + const int mo = t.tm_mon + 1; + const int d = t.tm_mday; + out[0] = static_cast('0' + (y / 1000) % 10); + out[1] = static_cast('0' + (y / 100) % 10); + out[2] = static_cast('0' + (y / 10) % 10); + out[3] = static_cast('0' + y % 10); + out[4] = '-'; + out[5] = static_cast('0' + mo / 10); + out[6] = static_cast('0' + mo % 10); + out[7] = '-'; + out[8] = static_cast('0' + d / 10); + out[9] = static_cast('0' + d % 10); + out[10] = '\0'; + return out; +} + +} // namespace + +bool window_open(const proto::Schedule& s, const std::tm& now) { + if (!s.enabled) return true; + + const int now_min = now.tm_hour * 60 + now.tm_min; + const std::optional start = s.startTime ? minutes_of(*s.startTime) : std::nullopt; + const std::optional stop = s.stopTime ? minutes_of(*s.stopTime) : std::nullopt; + + if (s.mode == proto::ScheduleMode::Once) { + if (!s.onceDate) return false; // "once" with no date never runs + const auto today = iso_date(now); + if (std::string_view(today.data()) != *s.onceDate) return false; + return in_window(now_min, start, stop); + } + + // periodic + if (s.daysOfWeek && !s.daysOfWeek->empty()) { + bool day_ok = false; + for (const auto d : *s.daysOfWeek) { + if (d == now.tm_wday) { + day_ok = true; + break; + } + } + if (!day_ok) return false; + } + return in_window(now_min, start, stop); +} + +bool window_open(const std::optional& s, const std::tm& now) { + if (!s) return true; + return window_open(*s, now); +} + +} // namespace velox::daemon::sched diff --git a/daemon/src/sched/schedule_window.hpp b/daemon/src/sched/schedule_window.hpp new file mode 100644 index 0000000..99aeeb4 --- /dev/null +++ b/daemon/src/sched/schedule_window.hpp @@ -0,0 +1,33 @@ +#pragma once + +// Evaluates a queue's Schedule against the current local wall-clock time: is the queue +// allowed to be running right now? Times are local HH:MM and re-evaluated on every tick +// (the daemon never caches an absolute instant — docs, Schedule.schema.json), so this is +// a pure function of (schedule, broken-down local time). +// +// A disabled schedule, or no schedule at all, means "always open" — the queue runs +// whenever queue.state is 'running'. + +#include + +#include "velox_proto.hpp" + +namespace velox::daemon::sched { + +// `local_now` must be a fully populated std::tm in local time (tm_wday and tm_year/mon/mday +// used). Returns true when the schedule permits the queue to run at that moment. +// +// - mode "once": open iff the date matches onceDate and the time is within the window. +// daysOfWeek is ignored. +// - mode "periodic": open iff today's weekday is in daysOfWeek (an empty/absent list means +// every day) and the time is within the window. +// +// The window is [startTime, stopTime). A null startTime means 00:00. A null stopTime +// means "until end of day" (and, for a queue, "until it drains"). A stopTime earlier than +// startTime is an overnight window (e.g. 22:00-06:00): open if now >= start OR now < stop. +bool window_open(const velox::proto::Schedule& schedule, const std::tm& local_now); + +// Convenience: the queue's schedule is optional. nullopt or disabled => always open. +bool window_open(const std::optional& schedule, const std::tm& local_now); + +} // namespace velox::daemon::sched diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index cd7269f..9301447 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -16,3 +16,5 @@ veloxd_test(store_migrations LIBS veloxd_store) veloxd_test(pairings LIBS veloxd_store veloxd_rpc) veloxd_test(ws_frame LIBS veloxd_rpc) veloxd_test(ws_server LIBS veloxd_rpc) +veloxd_test(sched_window LIBS veloxd_sched) +veloxd_test(sched_governor LIBS veloxd_sched) diff --git a/daemon/tests/sched_governor_test.cpp b/daemon/tests/sched_governor_test.cpp new file mode 100644 index 0000000..a706591 --- /dev/null +++ b/daemon/tests/sched_governor_test.cpp @@ -0,0 +1,178 @@ +#include "sched/governor.hpp" + +#include +#include + +#include "check.hpp" + +using namespace velox::daemon::sched; + +namespace { + +bool has(const std::vector& v, const std::string& x) { + return std::find(v.begin(), v.end(), x) != v.end(); +} + +TaskView task(std::string id, RunState st, std::optional queue = std::nullopt, + std::int64_t pos = 0, std::string host = "", std::int64_t rank = 0) { + TaskView t; + t.task_id = std::move(id); + t.run_state = st; + t.queue_id = std::move(queue); + t.queue_position = pos; + t.host = std::move(host); + t.admit_rank = rank; + return t; +} + +QueueView queue(std::string id, bool running, std::int64_t cap, bool window = true) { + QueueView q; + q.queue_id = std::move(id); + q.running = running; + q.max_concurrent = cap; + q.window_open = window; + return q; +} + +} // namespace + +void run() { + // --- global cap: 5 queued, max 3 -> start 3, order is all 3 -------------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 3, .max_active_segments = 32}); + std::vector t; + for (int i = 0; i < 5; ++i) + t.push_back(task("t" + std::to_string(i), RunState::Queued, std::nullopt, 0, "", i)); + const auto d = g.evaluate(t, {}); + CHECK_EQ(d.to_start.size(), 3u); + CHECK(has(d.to_start, "t0") && has(d.to_start, "t1") && has(d.to_start, "t2")); + CHECK_EQ(d.priority_order.size(), 3u); + CHECK_EQ(d.priority_order.front(), std::string("t0")); // FIFO by admit_rank + CHECK(d.to_pause.empty()); + } + + // --- the ADR 0011 §2 clamp: maxActiveSegments below maxConcurrentDownloads ------ + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 2}); + std::vector t; + for (int i = 0; i < 6; ++i) t.push_back(task("t" + std::to_string(i), RunState::Queued)); + const auto d = g.evaluate(t, {}); + CHECK_EQ(d.to_start.size(), 2u); // clamped to the segment budget + } + + // --- per-queue cap: queue running, cap 2, 4 queued in it ----------------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + std::vector t; + for (int i = 0; i < 4; ++i) + t.push_back(task("q" + std::to_string(i), RunState::Queued, "Q", i)); + const auto d = g.evaluate(t, {queue("Q", true, 2)}); + CHECK_EQ(d.to_start.size(), 2u); + CHECK(has(d.to_start, "q0") && has(d.to_start, "q1")); // lowest queue_position first + } + + // --- a stopped queue: its running tasks are paused (QueueStopped) ------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + std::vector t{ + task("r0", RunState::Running, "Q", 0), + task("r1", RunState::Running, "Q", 1), + task("m0", RunState::Running, std::nullopt, 0, "", 5), // main list, unaffected + }; + const auto d = g.evaluate(t, {queue("Q", /*running=*/false, 4)}); + CHECK_EQ(d.to_pause.size(), 2u); + CHECK(has(d.to_pause, "r0") && has(d.to_pause, "r1")); + CHECK(d.pause_reasons.at("r0") == PauseReason::QueueStopped); + CHECK(!has(d.to_pause, "m0")); + } + + // --- schedule window closed: pause reason is Schedule, not QueueStopped ------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + std::vector t{task("r0", RunState::Running, "Q", 0)}; + const auto d = g.evaluate(t, {queue("Q", /*running=*/true, 4, /*window=*/false)}); + CHECK_EQ(d.to_pause.size(), 1u); + CHECK(d.pause_reasons.at("r0") == PauseReason::Schedule); + } + + // --- resume: a task paused for Schedule comes back when the window reopens ---- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + TaskView p = task("p0", RunState::Paused, "Q", 0); + p.pause_reason = PauseReason::Schedule; + TaskView u = task("u0", RunState::Paused, "Q", 1); + u.pause_reason = PauseReason::User; // must NOT be auto-resumed + const auto d = g.evaluate({p, u}, {queue("Q", true, 4, true)}); + CHECK_EQ(d.to_resume.size(), 1u); + CHECK(has(d.to_resume, "p0")); + CHECK(!has(d.to_resume, "u0")); + } + + // --- an Auto (auth) pause is never touched, even with slots free ------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + TaskView a = task("a0", RunState::Paused, std::nullopt); + a.pause_reason = PauseReason::Auto; + const auto d = g.evaluate({a}, {}); + CHECK(d.to_resume.empty()); + CHECK(d.to_pause.empty()); + } + + // --- per-host cap: 3 queued on the same host, cap 1 ------------------------ + { + GovernorConfig cfg{.max_concurrent_downloads = 10, .max_active_segments = 32}; + cfg.host_caps["cdn.example"] = 1; + Governor g(cfg); + std::vector t{ + task("h0", RunState::Queued, std::nullopt, 0, "cdn.example", 0), + task("h1", RunState::Queued, std::nullopt, 0, "cdn.example", 1), + task("h2", RunState::Queued, std::nullopt, 0, "other.example", 2), + }; + const auto d = g.evaluate(t, {}); + CHECK_EQ(d.to_start.size(), 2u); // one per host + CHECK(has(d.to_start, "h0") && has(d.to_start, "h2")); + CHECK(!has(d.to_start, "h1")); + } + + // --- admission reconcile: 4 running, cap drops to 2 -> pause the 2 lowest ----- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 2, .max_active_segments = 32}); + std::vector t; + for (int i = 0; i < 4; ++i) + t.push_back(task("t" + std::to_string(i), RunState::Running, std::nullopt, 0, "", i)); + const auto d = g.evaluate(t, {}); + CHECK_EQ(d.to_pause.size(), 2u); + CHECK(has(d.to_pause, "t2") && has(d.to_pause, "t3")); // lowest priority (highest rank) + CHECK(d.pause_reasons.at("t2") == PauseReason::AdmissionReconcile); + CHECK_EQ(d.priority_order.size(), 2u); + } + + // --- determinism: shuffled input, identical Decision ----------------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 2, .max_active_segments = 32}); + std::vector a{ + task("b", RunState::Queued, std::nullopt, 0, "", 1), + task("a", RunState::Queued, std::nullopt, 0, "", 0), + task("c", RunState::Queued, std::nullopt, 0, "", 2), + }; + std::vector b{a[2], a[0], a[1]}; + const auto da = g.evaluate(a, {}); + const auto db = g.evaluate(b, {}); + CHECK(da.to_start == db.to_start); + CHECK(da.priority_order == db.priority_order); + CHECK_EQ(da.priority_order.front(), std::string("a")); + } + + // --- main list outranks queues in priority_order ------------------------- + { + Governor g(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32}); + std::vector t{ + task("qtask", RunState::Running, "Q", 0), + task("mtask", RunState::Running, std::nullopt, 0, "", 99), + }; + const auto d = g.evaluate(t, {queue("Q", true, 4)}); + CHECK_EQ(d.priority_order.front(), std::string("mtask")); + } +} + +TEST_MAIN() diff --git a/daemon/tests/sched_window_test.cpp b/daemon/tests/sched_window_test.cpp new file mode 100644 index 0000000..12223bd --- /dev/null +++ b/daemon/tests/sched_window_test.cpp @@ -0,0 +1,120 @@ +#include "sched/schedule_window.hpp" + +#include +#include + +#include "check.hpp" + +namespace proto = velox::proto; +using velox::daemon::sched::window_open; + +namespace { + +// A std::tm for a given weekday (0=Sun) and HH:MM on 2026-09-14 (a Monday) + offset days. +std::tm at(int wday, int hh, int mm, const char* date = "2026-09-14") { + std::tm t{}; + t.tm_year = 126; // 2026 + // parse "YYYY-MM-DD" + int y = 0, mo = 0, d = 0; + std::sscanf(date, "%d-%d-%d", &y, &mo, &d); + t.tm_year = y - 1900; + t.tm_mon = mo - 1; + t.tm_mday = d; + t.tm_wday = wday; + t.tm_hour = hh; + t.tm_min = mm; + return t; +} + +proto::Schedule periodic(const char* start, const char* stop, std::vector days) { + proto::Schedule s; + s.enabled = true; + s.mode = proto::ScheduleMode::Periodic; + if (start) s.startTime = start; + if (stop) s.stopTime = stop; + if (!days.empty()) s.daysOfWeek = std::move(days); + return s; +} + +} // namespace + +void run() { + // --- disabled schedule: always open -------------------------------------------- + { + proto::Schedule s; + s.enabled = false; + s.mode = proto::ScheduleMode::Periodic; + CHECK(window_open(s, at(3, 3, 0))); + std::optional none; + CHECK(window_open(none, at(3, 3, 0))); + } + + // --- periodic, weekdays 01:00-06:00 ------------------------------------------- + { + const auto s = periodic("01:00", "06:00", {1, 2, 3, 4, 5}); // Mon-Fri + CHECK(window_open(s, at(1, 2, 30))); // Monday 02:30 -> open + CHECK(!window_open(s, at(1, 6, 0))); // 06:00 is the exclusive end + CHECK(!window_open(s, at(1, 0, 59))); // before start + CHECK(!window_open(s, at(0, 2, 30))); // Sunday -> wrong day + CHECK(!window_open(s, at(6, 2, 30))); // Saturday -> wrong day + } + + // --- periodic, no daysOfWeek -> every day ------------------------------------ + { + const auto s = periodic("09:00", "17:00", {}); + CHECK(window_open(s, at(0, 12, 0))); // Sunday still fine + CHECK(window_open(s, at(3, 9, 0))); // inclusive start + CHECK(!window_open(s, at(3, 17, 0))); + } + + // --- overnight window 22:00-06:00 ------------------------------------------ + { + const auto s = periodic("22:00", "06:00", {}); + CHECK(window_open(s, at(3, 23, 0))); // late evening + CHECK(window_open(s, at(3, 2, 0))); // early morning + CHECK(!window_open(s, at(3, 12, 0))); // midday -> closed + CHECK(window_open(s, at(3, 22, 0))); // inclusive start + CHECK(!window_open(s, at(3, 6, 0))); // exclusive end + } + + // --- null stopTime -> until end of day ---------------------------------- + { + auto s = periodic("20:00", nullptr, {}); + CHECK(!window_open(s, at(3, 19, 59))); + CHECK(window_open(s, at(3, 20, 0))); + CHECK(window_open(s, at(3, 23, 59))); + } + + // --- null startTime -> from midnight ----------------------------------- + { + auto s = periodic(nullptr, "08:00", {}); + CHECK(window_open(s, at(3, 0, 0))); + CHECK(window_open(s, at(3, 7, 59))); + CHECK(!window_open(s, at(3, 8, 0))); + } + + // --- mode "once": only on the named date ------------------------------ + { + proto::Schedule s; + s.enabled = true; + s.mode = proto::ScheduleMode::Once; + s.startTime = "10:00"; + s.stopTime = "12:00"; + s.onceDate = "2026-09-14"; + CHECK(window_open(s, at(1, 11, 0, "2026-09-14"))); // right date + time + CHECK(!window_open(s, at(1, 9, 0, "2026-09-14"))); // right date, before window + CHECK(!window_open(s, at(2, 11, 0, "2026-09-15"))); // wrong date + s.daysOfWeek = {2}; // ignored for "once" + CHECK(window_open(s, at(1, 11, 0, "2026-09-14"))); + } + + // --- "once" with no onceDate never runs ------------------------------ + { + proto::Schedule s; + s.enabled = true; + s.mode = proto::ScheduleMode::Once; + CHECK(!window_open(s, at(1, 11, 0))); + } +} + +TEST_MAIN()