diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index cccadd2..4b01084 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -88,8 +88,16 @@ add_library(velox::daemon_rpc ALIAS veloxd_rpc) target_include_directories(veloxd_rpc PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src) target_compile_features(veloxd_rpc PUBLIC cxx_std_23) target_compile_options(veloxd_rpc PRIVATE -Wall -Wextra -Wpedantic -Werror) +# velox::core: dispatcher.hpp includes sched/scheduler.hpp for the Scheduler* it drives +# download.pause/resume/start/cancel and queue.start/stop through (D4b), which pulls in +# core/include's vdm/*.hpp. Interface-only from here (no .cpp in this library calls into +# CORE directly) — the actual Scheduler symbols resolve at the veloxd executable's link +# step (veloxd links both veloxd_rpc and veloxd_sched), not here, so this does not create +# the veloxd_rpc <-> veloxd_sched cycle that linking veloxd_sched itself would (veloxd_sched +# already links veloxd_rpc, for EventHub). target_link_libraries(veloxd_rpc - PUBLIC velox::proto veloxd_store veloxd_fs nlohmann_json::nlohmann_json Threads::Threads + PUBLIC velox::proto velox::core veloxd_store veloxd_fs nlohmann_json::nlohmann_json + Threads::Threads ) # --- veloxd — the daemon binary ------------------------------------------------------- diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 5208db4..52aad48 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -7,9 +7,9 @@ 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 the rest: `download.pause/resume/start/cancel/remove/addBatch/refreshUrl/provideAuth`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder/start/stop`, `category.upsert/remove`, `grabber.*`, `media.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list` and `queue.list` are done (`store/categories`, `store/queues`) | Per method, as each wires to the store/scheduler | +| D3 | Stub handlers for the rest: `download.remove/addBatch/refreshUrl/provideAuth/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder`, `category.upsert/remove`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list`, `queue.list`, `queue.start`/`stop` are done | Per method, as each wires to the store/scheduler | | ~~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 | +| ~~D4b~~ | **Closed** — `download.pause`/`resume`/`start`/`cancel` and `queue.start`/`stop` all drive the scheduler now, and apply *immediately* (not deferred to the next tick — pausing/resuming/cancelling a live transfer can't wait up to 1s, and per ADR 0013 §3 the governor never touches a user-owned pause on its own). New `rpc::TaskActionPort` interface (owned by `rpc/`, implemented by `sched::Scheduler`) is the seam dispatcher.hpp depends on instead of `sched/scheduler.hpp` directly — avoids a real `veloxd_rpc` <-> `veloxd_sched` circular library dependency (`veloxd_sched` already links `veloxd_rpc` for `EventHub`). `Scheduler::user_pause/resume/start/cancel` + `pause_queue` engine-call-then-eager-transition, matching `tick()`'s existing `to_pause` pattern. Fixed a real bug hit while building this: `transition()` always overwrote `pause_reason` to NULL when the engine's own delayed pause-ack callback arrived with no explicit reason, clobbering whatever the actual initiator (user or governor) had just written — now it preserves the stored reason when none is supplied. Verified against real `veloxd` + `tools/testserver`: pausing a live single-segment throttled transfer freezes `downloadedBytes`, resume continues it from that point, cancel stops it; `queue.stop(pauseRunning:true)` pauses the queue's running task immediately. NOTE: `download.start`'s contract "a task in 'queued' jumps its queue" (priority bump) is not implemented — admission is still plain FIFO by `created_at`. | `sched/scheduler.{cpp,hpp}`, `rpc/task_action_port.hpp`, `rpc/dispatcher.{hpp,cpp}`, `store/queues.{cpp,hpp}` | — | done, except the queue-jump priority bump noted above | | ~~D5~~ | **Mostly closed** — `rpc/event_hub` fans out per-subscription; `session.subscribe` on both transports registers/updates/tears down a real subscription; `Scheduler::transition()` publishes `event.task.state` (with `previousState`) on every state change, scheduler-driven or engine-reported; `dispatcher::on_download_add` publishes `event.task.added`; a 250 ms timer batches `Scheduler::progress_snapshot()` into one `event.task.progress` array per AGENT-DAEMON.md item 5 / the schema's `x-maxRateHz: 4`. Verified live end to end. | — | `event.task.removed` has no source yet (`download.remove` is D3); `event.speed.global`, `event.notify`, `event.auth.required`, `event.settings.changed`, `event.grabber.progress` are unpublished — each lands with its owning handler | as each owning D3 handler lands | | ~~D6~~ | **Closed** — engine numbers now reach the store: `Scheduler::tick()` probes (`EnginePort::probe`) before every `start()`, persisting `sizeBytes`/`resumable`/validators via `Tasks::set_probe_result` before a byte moves; `Scheduler::persist_progress()` (called from `progress_snapshot()` *and* once more from `on_engine_state` right before `release()`/unmap on every terminal transition) writes `downloadedBytes`/`speedBps`/`segments`/`segmentDetail` from the engine's `Progress`, so a task that finishes between two 250 ms ticks (the common case for anything small or fast) still leaves real numbers instead of the pre-persistence defaults. `TaskSummary.segments` is sourced from `segments.size()` when the task has any (matching what actually lands in `segmentDetail`, per the schema's "exactly `segments` entries"), falling back to the engine's `effective_segments` (budget slots *held*, not necessarily physical range count — see `core/include/vdm/task/download.hpp`'s `Progress` comment) only pre-segmentation. `Tasks::set_final_bytes` tops up `on_finished`'s byte count as a last-resort backstop. Migration `0002` adds `speed_bps` to both `tasks` and `segments`, and fixes `segments.state`'s CHECK to include `'pending'` (0001 omitted it, so a pre-connect snapshot could never be written). Verified against real `veloxd` + `tools/testserver` (not just unit tests): `download.list`/`download.get` correct immediately after completion and after a daemon restart. | `sched/scheduler.{cpp,hpp}`, `store/{tasks,segments}.{cpp,hpp}`, `store/migrations/0002_*.sql` | — | done | -| — | **Observed, not fixed (CORE, not this lane):** `vdm::task::Progress.speed_bps` reads back as `0` for the whole lifetime of a live, real (non-fake) throttled download in the E2E check above, despite `downloadedBytes` visibly advancing between polls — `core/src/task/download_task.cpp`'s per-worker EWMA (`w->speed_bps`, ~line 505-513) never seems to produce a nonzero aggregate in this build. DAEMON passes `EnginePort::progress()`'s `speed_bps` straight through (`Scheduler::persist_progress`); nothing in this lane drops it. Filed here rather than worked around — CLAUDE.md §2/§3: not core/'s owner, don't patch around a wrong upstream number locally. Confirm with CORE before the GUI's live speed readout ships. | +| — | ~~Observed, not fixed (CORE, not this lane)~~ — **routed to CORE by the user.** `vdm::task::Progress.speed_bps` reads back as `0` for the whole lifetime of a live, real (non-fake) throttled download, despite `downloadedBytes` visibly advancing between polls — `core/src/task/download_task.cpp`'s per-worker EWMA never seems to produce a nonzero aggregate in this build. DAEMON passes `EnginePort::progress()`'s `speed_bps` straight through (`Scheduler::persist_progress`); nothing in this lane drops it. Still reproduces in the D4b live checks above (0 throughout a paused/resumed/cancelled transfer whose `downloadedBytes` visibly moved) — not re-filed, since it's already CORE's. | diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index a70a9bf..788ce4a 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -103,7 +103,7 @@ int main() { (void)scheduler.reload_config(); (void)scheduler.tick(); // admit anything already queued in the DB - velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub); + velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub, &scheduler); dispatcher.set_on_mutation([&loop, &scheduler] { loop.post([&scheduler] { (void)scheduler.tick(); }); }); diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index fafec40..9f11f5d 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -28,6 +28,29 @@ proto::HandlerResult not_implemented(const char* method) { std::string("not implemented in this build: ") + method}); } +// Shared by download.pause/resume/start/cancel: apply `action` to every id in `task_ids` +// and fold the per-task velox::daemon::rpc::TaskActionPort::Result into a +// BulkTaskResult (ADR: "a bulk call never fails as a whole because one id was bad"). +proto::BulkTaskResult bulk_apply( + const std::vector& task_ids, + const std::function& + action) { + proto::BulkTaskResult out; + for (const auto& id : task_ids) { + const auto r = action(id); + if (!r.found) { + out.failed.push_back({id, proto::ErrorCode::TaskNotFound, "no such task"}); + continue; + } + proto::BulkTaskResultUpdatedItem item; + item.taskId = id; + item.changed = r.changed; + if (auto st = proto::parse_TaskState(r.state)) item.state = *st; + out.updated.push_back(std::move(item)); + } + return out; +} + // A v4 UUID for a new task id. std::string new_task_id() { std::random_device rd; @@ -237,8 +260,13 @@ VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) { return not_implemented("download.addBatch"); } proto::HandlerResult -VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) { - return not_implemented("download.cancel"); +VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) { + if (!actions_) return not_implemented("download.cancel"); + auto result = bulk_apply(params.taskIds, [this](const std::string& id) { + return actions_->user_cancel(id, /*discard_partial=*/false); + }); + if (on_mutation_) on_mutation_(); + return result; } proto::HandlerResult VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) { @@ -273,8 +301,12 @@ VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) { return d; } proto::HandlerResult -VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) { - return not_implemented("download.pause"); +VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams& params) { + if (!actions_) return not_implemented("download.pause"); + auto result = bulk_apply(params.taskIds, + [this](const std::string& id) { return actions_->user_pause(id); }); + if (on_mutation_) on_mutation_(); + return result; } proto::HandlerResult VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) { @@ -293,12 +325,20 @@ VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) { return not_implemented("download.remove"); } proto::HandlerResult -VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams&) { - return not_implemented("download.resume"); +VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) { + if (!actions_) return not_implemented("download.resume"); + auto result = bulk_apply( + params.taskIds, [this](const std::string& id) { return actions_->user_resume(id); }); + if (on_mutation_) on_mutation_(); + return result; } proto::HandlerResult -VeloxDispatcher::on_download_start(const proto::DownloadStartParams&) { - return not_implemented("download.start"); +VeloxDispatcher::on_download_start(const proto::DownloadStartParams& params) { + if (!actions_) return not_implemented("download.start"); + auto result = bulk_apply( + params.taskIds, [this](const std::string& id) { return actions_->user_start(id); }); + if (on_mutation_) on_mutation_(); + return result; } proto::HandlerResult VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams&) { @@ -346,12 +386,54 @@ VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) { return not_implemented("queue.reorder"); } proto::HandlerResult -VeloxDispatcher::on_queue_start(const proto::QueueStartParams&) { - return not_implemented("queue.start"); +VeloxDispatcher::on_queue_start(const proto::QueueStartParams& params) { + store::Queues queues(db_); + auto exists = queues.get(params.queueId); + if (!exists) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.start: " + exists.error().message}); + if (!exists->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such queue", + nlohmann::json{{"queueId", params.queueId}}}); + + (void)queues.set_state(params.queueId, "running"); + // Nudged through the same deferred tick() as download.add (on_mutation_ posts onto the + // loop) — which tasks actually get admitted happens asynchronously (tick()'s to_start + // even probes before it starts one), so startedTaskIds is not knowable synchronously + // here; the GUI learns the real outcome from event.task.state as each one lands. + if (on_mutation_) on_mutation_(); + + auto after = queues.get(params.queueId); + proto::QueueStartResult r; + if (after && after->has_value()) r.queue = **after; + return r; } proto::HandlerResult -VeloxDispatcher::on_queue_stop(const proto::QueueStopParams&) { - return not_implemented("queue.stop"); +VeloxDispatcher::on_queue_stop(const proto::QueueStopParams& params) { + store::Queues queues(db_); + auto exists = queues.get(params.queueId); + if (!exists) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.stop: " + exists.error().message}); + if (!exists->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such queue", + nlohmann::json{{"queueId", params.queueId}}}); + + (void)queues.set_state(params.queueId, "stopped"); + + proto::QueueStopResult r; + // "the difference between 'stop the queue' and 'stop everything', which IDM + // conflates" (the schema's own words) — absent means the soft stop: only halt new + // admissions, let what's running finish. + if (params.pauseRunning.value_or(false) && actions_) + r.pausedTaskIds = actions_->pause_queue(params.queueId); + if (on_mutation_) on_mutation_(); + + auto after = queues.get(params.queueId); + if (after && after->has_value()) r.queue = **after; + return r; } proto::HandlerResult VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) { diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 126b59c..2c4ccc5 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -15,6 +15,7 @@ #include #include "rpc/event_hub.hpp" +#include "rpc/task_action_port.hpp" #include "store/sqlite.hpp" #include "velox_proto.hpp" @@ -22,7 +23,15 @@ namespace velox::daemon::rpc { class VeloxDispatcher final : public velox::proto::Dispatcher { public: - VeloxDispatcher(velox::daemon::store::Db& db, EventHub& hub) : db_(db), hub_(hub) {} + // `actions` drives download.pause/resume/start/cancel and queue.start/stop + // immediately (D4b) — those cannot wait for the next tick(), unlike download.add's + // on_mutation nudge. Optional so existing tests that only exercise download.add/list/ + // get keep building with no scheduler at hand; a null actions_ makes those methods + // answer "not implemented" instead of crashing. See rpc/task_action_port.hpp for why + // this is an interface owned by rpc/ rather than a direct sched::Scheduler* (avoids a + // veloxd_rpc <-> veloxd_sched circular library dependency). + VeloxDispatcher(velox::daemon::store::Db& db, EventHub& hub, TaskActionPort* actions = nullptr) + : db_(db), hub_(hub), actions_(actions) {} // Called after a handler mutates task state (download.add for now). main.cpp wires it // to nudge the scheduler; unset in tests. @@ -109,6 +118,7 @@ public: private: velox::daemon::store::Db& db_; EventHub& hub_; + TaskActionPort* actions_; std::function on_mutation_; }; diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp new file mode 100644 index 0000000..6f03218 --- /dev/null +++ b/daemon/src/rpc/task_action_port.hpp @@ -0,0 +1,44 @@ +#pragma once + +// The seam between the dispatcher and the scheduler for user-initiated task/queue actions +// (download.pause/resume/start/cancel, queue.stop's pauseRunning) — owned by rpc/ so +// dispatcher.hpp (part of veloxd_rpc) never has to include sched/scheduler.hpp, which +// would make veloxd_rpc depend on veloxd_sched at compile time. veloxd_sched already +// depends on veloxd_rpc (for EventHub); the other direction too would be a real circular +// library dependency, not just an inconvenience — anything linking veloxd_rpc alone (e.g. +// the CLI's tests) would fail to link over symbols it never calls. +// +// sched::Scheduler implements this directly (it already lives in a library that depends on +// rpc/, so adding an rpc-defined base costs nothing new); main.cpp hands the dispatcher a +// `TaskActionPort*` pointing at the same Scheduler it constructs. + +#include +#include + +namespace velox::daemon::rpc { + +class TaskActionPort { +public: + virtual ~TaskActionPort() = default; + + // Mirrors sched::Scheduler::UserActionResult: whether the task was found at all, + // whether it actually changed state (a task already in the target/a terminal state is + // reported found=true, changed=false — BulkTaskResult's own "not an error" contract), + // and its resulting/current state spelling either way. + struct Result { + bool found = false; + bool changed = false; + std::string state; + }; + + virtual Result user_pause(const std::string& wire_id) = 0; + virtual Result user_resume(const std::string& wire_id) = 0; + virtual Result user_start(const std::string& wire_id) = 0; + virtual Result user_cancel(const std::string& wire_id, bool discard_partial) = 0; + + // queue.stop(pauseRunning=true): pause every currently-running task in `queue_id` now. + // Returns the wire ids actually paused. + virtual std::vector pause_queue(const std::string& queue_id) = 0; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index 52a26ba..f460a53 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -291,7 +291,16 @@ void Scheduler::transition(const std::string& wire_id, std::string_view to_state // An engine-initiated pause carries an error => 'auto' (ADR 0013 §2), overriding // whatever the caller passed (a scheduler-driven pause never carries an error here). std::optional reason = pause_reason; - if (to_state == "paused" && err) reason = "auto"; + if (to_state == "paused" && err) { + reason = "auto"; + } else if (to_state == "paused" && !reason && before && before->has_value()) { + // No reason supplied — the common case is the engine's own pause-ack callback + // (on_state(_, paused, nullopt)) arriving after whoever actually initiated the + // pause (user_pause() or tick()'s to_pause loop) already wrote the real reason + // eagerly. Keep what's already stored instead of clobbering it back to NULL: + // set_state() always overwrites the column, reason or not. + reason = (**before).pause_reason; + } (void)tasks.set_state(wire_id, to_state, reason); if (err) { @@ -472,4 +481,98 @@ std::vector Scheduler::progress_snapshot() { return out; } +namespace { +bool is_terminal_state(const std::string& s) { + return s == "complete" || s == "failed" || s == "cancelled"; +} +} // namespace + +rpc::TaskActionPort::Result Scheduler::user_pause(const std::string& wire_id) { + store::Tasks tasks(db_); + auto got = tasks.get(wire_id); + if (!got || !got->has_value()) return {false, false, {}}; + const store::TaskRow& row = **got; + if (is_terminal_state(row.state) || row.state == "paused") + return {true, false, row.state}; + + if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid); + transition(wire_id, "paused", std::string("user"), std::nullopt); + return {true, true, "paused"}; +} + +rpc::TaskActionPort::Result Scheduler::user_resume(const std::string& wire_id) { + store::Tasks tasks(db_); + auto got = tasks.get(wire_id); + if (!got || !got->has_value()) return {false, false, {}}; + const store::TaskRow& row = **got; + if (row.state != "paused") return {true, false, row.state}; + + if (auto eid = engine_id_of(wire_id)) { + engine_.resume(*eid); + transition(wire_id, "connecting", std::nullopt, std::nullopt); + return {true, true, "connecting"}; + } + transition(wire_id, "queued", std::nullopt, std::nullopt); + return {true, true, "queued"}; +} + +rpc::TaskActionPort::Result Scheduler::user_start(const std::string& wire_id) { + store::Tasks tasks(db_); + auto got = tasks.get(wire_id); + if (!got || !got->has_value()) return {false, false, {}}; + const store::TaskRow& row = **got; + if (row.state != "paused" && row.state != "new") return {true, false, row.state}; + + if (auto eid = engine_id_of(wire_id)) { + engine_.resume(*eid); + transition(wire_id, "connecting", std::nullopt, std::nullopt); + return {true, true, "connecting"}; + } + transition(wire_id, "queued", std::nullopt, std::nullopt); + return {true, true, "queued"}; +} + +rpc::TaskActionPort::Result Scheduler::user_cancel(const std::string& wire_id, + bool discard_partial) { + store::Tasks tasks(db_); + auto got = tasks.get(wire_id); + if (!got || !got->has_value()) return {false, false, {}}; + const store::TaskRow& row = **got; + if (is_terminal_state(row.state)) return {true, false, row.state}; + + // Same pattern as tick()'s to_pause loop: call the engine (async, no synchronous + // effect) and transition the store eagerly so download.get/list are correct the + // instant this call returns. The engine's own on_state(_, cancelled, nullopt) + + // on_finished arrive later via on_engine_state, which is what actually + // release()s/unmaps the handle — never done here. + if (auto eid = engine_id_of(wire_id)) engine_.cancel(*eid, discard_partial); + transition(wire_id, "cancelled", std::nullopt, std::nullopt); + return {true, true, "cancelled"}; +} + +std::vector Scheduler::pause_queue(const std::string& queue_id) { + store::Tasks tasks(db_); + proto::TaskFilter filter; + filter.queueId = queue_id; + filter.states = non_terminal_states(); + // No paging needed: a queue's max_concurrent is <= 32, so "everything non-terminal in + // this queue" is never a large page. + auto page = tasks.list(filter, std::nullopt, 0, 10000); + std::vector paused; + if (!page) return paused; + + for (const auto& row : page->rows) { + if (run_state_of(row.state) != RunState::Running) continue; + if (auto eid = engine_id_of(row.task_id)) engine_.pause(*eid); + transition(row.task_id, "paused", std::string("queue_stopped"), std::nullopt); + paused.push_back(row.task_id); + } + return paused; +} + +void Scheduler::probe_now(const vdm::net::ProbeRequest& req, + std::function)> done) { + engine_.probe(req, std::move(done)); +} + } // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index c4ba73d..d9db9ae 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -25,6 +25,7 @@ #include #include "rpc/event_hub.hpp" +#include "rpc/task_action_port.hpp" #include "sched/engine_port.hpp" #include "sched/governor.hpp" #include "store/sqlite.hpp" @@ -45,7 +46,12 @@ struct TaskErrorFields { std::optional attempt; }; -class Scheduler { +// Implements rpc::TaskActionPort directly — sched/ already depends on rpc/ (EventHub), so +// this costs nothing new, and it's what lets dispatcher.hpp depend on the port interface +// instead of on sched/scheduler.hpp (see rpc/task_action_port.hpp's top comment for why +// that matters: it would otherwise make veloxd_rpc <-> veloxd_sched a circular library +// dependency). +class Scheduler final : public rpc::TaskActionPort { 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 @@ -98,6 +104,42 @@ public: }; std::vector progress_snapshot(); + // rpc::TaskActionPort. These apply immediately — never wait for the next tick() — + // because pausing, resuming or cancelling a live transfer cannot wait up to 1s for the + // timerfd, and the governor will never do any of them on its own for a user-owned + // reason (ADR 0013 §3: "never touch a task paused for a reason it does not own"). + // Idempotent: calling one on a task already in the target (or a terminal) state + // reports found=true, changed=false. + rpc::TaskActionPort::Result user_pause(const std::string& wire_id) override; + // A task still holding a live engine handle (paused mid-flight) is engine_.resume()'d + // straight back to `connecting`; one with no handle yet (parked since download.add + // with startMode 'later', or never admitted) goes to `queued` for the next tick's + // normal admission. + rpc::TaskActionPort::Result user_resume(const std::string& wire_id) override; + // "Begin or restart the given tasks" (download.start): same effect as user_resume for + // a paused/new task. NOTE: the contract's "a task in 'queued' jumps its queue" priority + // bump is not implemented — admission is still plain FIFO via the governor's + // created_at rank. Flagged in deferrals.md. + rpc::TaskActionPort::Result user_start(const std::string& wire_id) override; + // download.cancel == cancel(discard_partial=false); download.remove == cancel(true) + // plus the store row / file cleanup (that part is still D3). + rpc::TaskActionPort::Result user_cancel(const std::string& wire_id, + bool discard_partial) override; + + // queue.stop(pauseRunning=true): pause every task in `queue_id` the governor would + // currently call Running, right now rather than waiting for the next tick — the same + // immediacy reasoning as the user_* actions above, with PauseReason::QueueStopped + // instead of User. Returns the wire ids actually paused. + std::vector pause_queue(const std::string& queue_id) override; + + // download.probe's standalone use (File Info dialog, no task row involved): a thin + // passthrough to the engine's own probe pool, outside the segment budget (ADR 0011 + // §5). Never blocks — `done` arrives on an engine thread like every other EnginePort + // callback; the caller (the RPC server layer, not this synchronous dispatcher — see + // rpc/dispatcher.hpp's top comment) is responsible for marshalling the reply back. + void probe_now(const vdm::net::ProbeRequest& req, + std::function)> done); + // Diagnostics / tests. std::optional wire_id_of(vdm::TaskId id) const; std::optional engine_id_of(const std::string& wire_id) const; diff --git a/daemon/src/store/queues.cpp b/daemon/src/store/queues.cpp index 97c63b8..9fb469f 100644 --- a/daemon/src/store/queues.cpp +++ b/daemon/src/store/queues.cpp @@ -1,11 +1,44 @@ #include "store/queues.hpp" +#include + #include namespace velox::daemon::store { namespace proto = velox::proto; +namespace { + +// One queue row (columns queue_id, name, state, max_concurrent, schedule, in that order) +// plus its member taskIds, read off the row a caller has already step()'d to. +DbResult project_row(Db& db, Stmt& st) { + proto::Queue q; + q.queueId = st.column_text(0); + q.name = st.column_text(1); + if (auto s = proto::parse_QueueState(st.column_text(2))) q.state = *s; + q.maxConcurrent = st.column_int(3); + if (!st.column_is_null(4)) { + auto j = nlohmann::json::parse(st.column_text(4), nullptr, false); + if (auto sched = proto::parse(j, "schedule")) q.schedule = *sched; + } + + auto ts = db.prepare("SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position"); + if (!ts) return std::unexpected(ts.error()); + if (auto b = ts->bind(1, std::string_view(q.queueId)); !b) return std::unexpected(b.error()); + std::vector ids; + for (;;) { + auto r = ts->step(); + if (!r) return std::unexpected(r.error()); + if (!*r) break; + ids.push_back(ts->column_text(0)); + } + q.taskIds = std::move(ids); + return q; +} + +} // namespace + DbResult> Queues::list() { auto st = db_.prepare( "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name"); @@ -16,33 +49,33 @@ DbResult> Queues::list() { auto row = st->step(); if (!row) return std::unexpected(row.error()); if (!*row) break; - - proto::Queue q; - q.queueId = st->column_text(0); - q.name = st->column_text(1); - if (auto s = proto::parse_QueueState(st->column_text(2))) q.state = *s; - q.maxConcurrent = st->column_int(3); - if (!st->column_is_null(4)) { - auto j = nlohmann::json::parse(st->column_text(4), nullptr, false); - if (auto sched = proto::parse(j, "schedule")) q.schedule = *sched; - } - - auto ts = db_.prepare( - "SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position"); - if (!ts) return std::unexpected(ts.error()); - if (auto b = ts->bind(1, std::string_view(q.queueId)); !b) return std::unexpected(b.error()); - std::vector ids; - for (;;) { - auto r = ts->step(); - if (!r) return std::unexpected(r.error()); - if (!*r) break; - ids.push_back(ts->column_text(0)); - } - q.taskIds = std::move(ids); - - out.push_back(std::move(q)); + auto q = project_row(db_, *st); + if (!q) return std::unexpected(q.error()); + out.push_back(std::move(*q)); } return out; } +DbResult> Queues::get(std::string_view queue_id) { + auto st = db_.prepare( + "SELECT queue_id, name, state, max_concurrent, schedule FROM queues WHERE queue_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error()); + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return std::optional{}; + auto q = project_row(db_, *st); + if (!q) return std::unexpected(q.error()); + return std::optional{std::move(*q)}; +} + +DbResult Queues::set_state(std::string_view queue_id, std::string_view state) { + auto st = db_.prepare("UPDATE queues SET state = ?2 WHERE queue_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error()); + if (auto b = st->bind(2, state); !b) return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + } // namespace velox::daemon::store diff --git a/daemon/src/store/queues.hpp b/daemon/src/store/queues.hpp index 90460bb..dd1c8ef 100644 --- a/daemon/src/store/queues.hpp +++ b/daemon/src/store/queues.hpp @@ -5,6 +5,8 @@ // — membership changes through download.update / queue.reorder, per Queue's own schema // note that a queue.upsert payload's taskIds is ignored. +#include +#include #include #include "store/sqlite.hpp" @@ -18,6 +20,13 @@ public: DbResult> list(); + // nullopt (not an error) if no queue has this id. + DbResult> get(std::string_view queue_id); + + // 'running' or 'stopped' (Queue.schema.json / the state column's CHECK). false if the + // id doesn't exist. + DbResult set_state(std::string_view queue_id, std::string_view state); + private: Db& db_; }; diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp index 3adf08b..11f281c 100644 --- a/daemon/tests/sched_scheduler_test.cpp +++ b/daemon/tests/sched_scheduler_test.cpp @@ -259,6 +259,108 @@ void run() { tasks.remove("pr0"); } + + // --- user_pause / user_resume: a live task, and one that never started ----------- + { + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + CHECK(tasks.insert(task("up0", "queued", "2026-09-10T10:00:00Z")).has_value()); + CHECK(tasks.insert(task("up1", "paused", "2026-09-10T10:00:00Z")).has_value()); + CHECK(sched.tick().has_value()); // admits up0; up1 stays paused (governor never + // touches a user-owned pause) + CHECK_EQ(engine.starts.size(), 1u); + const auto live_id = engine.starts[0].id; + + // Pausing a live task calls the engine now and transitions eagerly — not left for + // the next tick. + auto r = sched.user_pause("up0"); + CHECK(r.found); + CHECK(r.changed); + CHECK_EQ(r.state, std::string("paused")); + CHECK_EQ(engine.paused.size(), 1u); + CHECK_EQ(engine.paused[0].value, live_id.value); + CHECK_EQ(task_state(*db, "up0"), std::string("paused")); + auto row = tasks.get("up0").value().value(); + CHECK_EQ(row.pause_reason.value_or(""), std::string("user")); + + // Idempotent: pausing an already-paused task is a no-op, not an error. + auto again = sched.user_pause("up0"); + CHECK(again.found); + CHECK(!again.changed); + + // Resuming a task that still holds a live engine handle calls engine.resume() and + // goes straight to `connecting`. + auto res = sched.user_resume("up0"); + CHECK(res.found); + CHECK(res.changed); + CHECK_EQ(res.state, std::string("connecting")); + CHECK_EQ(engine.resumed.size(), 1u); + CHECK_EQ(engine.resumed[0].value, live_id.value); + + // The engine's own delayed pause-ack (on_state with no error, arriving after the + // eager transition already wrote the real reason) must not clobber pause_reason + // back to NULL. + (void)sched.user_pause("up0"); + sched.on_engine_state("up0", "downloading", "paused", std::nullopt); + auto row2 = tasks.get("up0").value().value(); + CHECK_EQ(row2.pause_reason.value_or(""), std::string("user")); + + // up1 never started (still parked, no engine handle): resume just re-queues it for + // the next tick's normal admission. + auto res2 = sched.user_resume("up1"); + CHECK(res2.found); + CHECK(res2.changed); + CHECK_EQ(res2.state, std::string("queued")); + CHECK(engine.resumed.size() == 1u); // up1 was never mapped; no engine call + + // Not found: a bogus id reports found=false, not a crash. + auto missing = sched.user_pause("does-not-exist"); + CHECK(!missing.found); + + tasks.remove("up0"); + tasks.remove("up1"); + } + + // --- user_cancel + pause_queue -------------------------------------------------- + { + 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("uc0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value()); + CHECK(tasks.insert(task("uc1", "queued", "2026-09-10T10:00:01Z", "main", 1)).has_value()); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 2u); + + auto c = sched.user_cancel("uc0", /*discard_partial=*/true); + CHECK(c.found); + CHECK(c.changed); + CHECK_EQ(c.state, std::string("cancelled")); + CHECK_EQ(engine.cancelled.size(), 1u); + CHECK(engine.cancelled[0].second); // discard_partial passed through + CHECK_EQ(task_state(*db, "uc0"), std::string("cancelled")); + + // Cancelling an already-terminal task is a no-op. + auto c2 = sched.user_cancel("uc0", false); + CHECK(c2.found); + CHECK(!c2.changed); + + // pause_queue pauses every still-running task in the queue (uc0 is terminal, so + // only uc1 is affected) and reports pause_reason 'queue_stopped'. + const auto paused_ids = sched.pause_queue("main"); + CHECK_EQ(paused_ids.size(), std::size_t{1}); + CHECK_EQ(paused_ids[0], std::string("uc1")); + CHECK_EQ(task_state(*db, "uc1"), std::string("paused")); + auto row = tasks.get("uc1").value().value(); + CHECK_EQ(row.pause_reason.value_or(""), std::string("queue_stopped")); + + tasks.remove("uc0"); + tasks.remove("uc1"); + CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value()); + } } TEST_MAIN()