From de748cc2fc714bbff3a03b518fc37225228b5887 Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:04:02 +0400 Subject: [PATCH 1/5] daemon: fix startMode 'later' -32603 and isolate the single-instance lock by runtime dir MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two bugs blocking PROTO's live-veloxd conformance check. 1. tasks.start_mode's CHECK was ('auto','now','queue','manual') — not the contract's StartMode enum (['now','later','queue']) at all. 'later', a real documented value (the File Info dialog's Download Later button), hit the CHECK on every insert and surfaced as an unhandled -32603; 'auto'/'manual' were never contract values to begin with. Migration 0003 rebuilds tasks (SQLite can't ALTER a CHECK) with the contract's values, remapping existing rows by what they actually meant: 'auto' -> 'now' (eligible for the scheduler immediately), 'manual' -> 'later' (parked, matching StartMode's own "lands the task in paused" description). store_migrations_test covers the remap and that 'later' inserts clean while the retired spellings are rejected. dispatcher.cpp's on_download_add matched: default (absent startMode) is now 'now' instead of the invented 'auto'; 'later' actually lands the task in `paused` (pause_reason 'user') instead of a dead 'manual' -> `new` branch that spec.startMode (typed as the 3-value enum) could never even reach. TaskRow::start_mode's in-memory default followed suit ('now'). 2. main.cpp's single-instance guard bound an abstract socket named "velox-daemon-" — one name per user, system-wide. XDG_RUNTIME_DIR isolation never reached it: a leaked test veloxd held the lock for 4h40m and locked out every other isolated instance with the same euid (PROTO, EXT, the orchestrator), real daemon included. Extracted rpc/single_instance.{hpp,cpp} (was a static in main.cpp, untestable) and derived the abstract-socket name from a hash of the resolved runtime dir path instead of euid alone. The real per-user daemon is still unique (its runtime dir is unique to it); isolated instances pointed at their own runtime dirs now coexist. main() resolves the runtime dir before acquiring the lock (was the other way around). New single_instance_test covers same-dir refusal, different-dir coexistence, and release-on-close. Verified against real veloxd binaries, not just unit tests: startMode: "later" via a live download.add lands in `paused`; two veloxd with different runtime dirs run concurrently, two with the same one and the second refuses with the runtime dir named in the error. Full ctest: 39/39. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/CMakeLists.txt | 1 + daemon/src/main.cpp | 38 ++------ daemon/src/rpc/dispatcher.cpp | 17 +++- daemon/src/rpc/single_instance.cpp | 38 ++++++++ daemon/src/rpc/single_instance.hpp | 27 ++++++ .../0003_start_mode_contract_values.sql | 91 +++++++++++++++++++ daemon/src/store/tasks.hpp | 2 +- daemon/tests/CMakeLists.txt | 1 + daemon/tests/single_instance_test.cpp | 40 ++++++++ daemon/tests/store_migrations_test.cpp | 50 ++++++++++ 10 files changed, 270 insertions(+), 35 deletions(-) create mode 100644 daemon/src/rpc/single_instance.cpp create mode 100644 daemon/src/rpc/single_instance.hpp create mode 100644 daemon/src/store/migrations/0003_start_mode_contract_values.sql create mode 100644 daemon/tests/single_instance_test.cpp diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index 6d896ea..cccadd2 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -73,6 +73,7 @@ target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------ add_library(veloxd_rpc STATIC src/rpc/runtime_dir.cpp + src/rpc/single_instance.cpp src/rpc/event_loop.cpp src/rpc/event_hub.cpp src/rpc/uds_server.cpp diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index 14a1567..a70a9bf 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -24,6 +24,7 @@ #include "rpc/event_loop.hpp" #include "rpc/pairing.hpp" #include "rpc/runtime_dir.hpp" +#include "rpc/single_instance.hpp" #include "rpc/uds_server.hpp" #include "rpc/ws_server.hpp" #include "sched/engine_port_core.hpp" @@ -43,48 +44,25 @@ void on_signal(int) { if (g_loop != nullptr) g_loop->stop(); // stop() is async-signal-safe (writes an eventfd) } -// Single-instance guard: bind an abstract-namespace Unix socket whose name is unique to -// this user. A second daemon gets EADDRINUSE and exits. The kernel reclaims an -// abstract-namespace address when the holding process dies, so a crash never wedges it -// (docs/01 §2). Returns the held fd (kept open for the process lifetime) or -1. -int acquire_single_instance_lock() { - const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); - if (fd < 0) return -1; - - const std::string name = std::string("velox-daemon-") + std::to_string(::geteuid()); - sockaddr_un addr{}; - addr.sun_family = AF_UNIX; - // Leading NUL selects the abstract namespace; the name follows, not NUL-terminated. - addr.sun_path[0] = '\0'; - std::memcpy(addr.sun_path + 1, name.c_str(), name.size()); - const socklen_t len = - static_cast(offsetof(sockaddr_un, sun_path) + 1 + name.size()); - - if (::bind(fd, reinterpret_cast(&addr), len) != 0) { - ::close(fd); - return -1; - } - return fd; -} - } // namespace int main() { std::cout << "veloxd " << velox::daemon::kDaemonVersion << " (protocol " << velox::proto::kProtocolVersion << ")\n"; - const int lock_fd = acquire_single_instance_lock(); - if (lock_fd < 0) { - std::cerr << "veloxd: another instance is already running for this user\n"; - return 1; - } - velox::daemon::rpc::RuntimeDir rt; if (const auto ec = velox::daemon::rpc::resolve_runtime_dir(rt)) { std::cerr << "veloxd: cannot prepare runtime directory: " << ec.message() << "\n"; return 1; } + const int lock_fd = velox::daemon::rpc::acquire_single_instance_lock(rt.path); + if (lock_fd < 0) { + std::cerr << "veloxd: another instance is already running for this runtime " + "directory (" << rt.path << ")\n"; + return 1; + } + velox::daemon::rpc::EventLoop loop; g_loop = &loop; diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 524ad37..fafec40 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -152,10 +152,19 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { row.save_dir = target->dir; row.filename = target->leaf; 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`); on_mutation_ nudges it. - row.state = row.start_mode == "manual" ? "new" : "queued"; + // Contract values only (StartMode.schema.json: 'now'|'later'|'queue'); absent means + // "start it", same as an explicit 'now'. + row.start_mode = + spec.startMode ? std::string(proto::to_string(*spec.startMode)) : "now"; + // 'later' — the File Info dialog's Download Later button — lands the task in `paused` + // per StartMode's own description; anything else ('now' or 'queue') is eligible for + // the scheduler immediately (`queued`); on_mutation_ nudges it. + if (row.start_mode == "later") { + row.state = "paused"; + row.pause_reason = "user"; + } else { + row.state = "queued"; + } row.category_id = spec.categoryId; row.queue_id = spec.queueId; row.description = spec.description; diff --git a/daemon/src/rpc/single_instance.cpp b/daemon/src/rpc/single_instance.cpp new file mode 100644 index 0000000..5385c96 --- /dev/null +++ b/daemon/src/rpc/single_instance.cpp @@ -0,0 +1,38 @@ +#include "rpc/single_instance.hpp" + +#include +#include + +#include +#include +#include + +#include "util/crypto.hpp" + +namespace velox::daemon::rpc { + +int acquire_single_instance_lock(const std::string& runtime_dir) { + const int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (fd < 0) return -1; + + // Truncated to 16 hex chars (64 bits): a collision would need two distinct runtime + // dirs to hash together, which is not a security boundary here — the socket itself is + // still 0600 same-UID-checked; this is only "don't let two daemons stomp each other". + const std::string name = + "velox-daemon-" + velox::daemon::crypto::sha256_hex(runtime_dir).substr(0, 16); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + // Leading NUL selects the abstract namespace; the name follows, not NUL-terminated. + addr.sun_path[0] = '\0'; + std::memcpy(addr.sun_path + 1, name.c_str(), name.size()); + const socklen_t len = + static_cast(offsetof(sockaddr_un, sun_path) + 1 + name.size()); + + if (::bind(fd, reinterpret_cast(&addr), len) != 0) { + ::close(fd); + return -1; + } + return fd; +} + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/single_instance.hpp b/daemon/src/rpc/single_instance.hpp new file mode 100644 index 0000000..c2b87a4 --- /dev/null +++ b/daemon/src/rpc/single_instance.hpp @@ -0,0 +1,27 @@ +#pragma once + +// Single-instance guard: bind an abstract-namespace Unix socket whose name is derived +// from the canonical runtime directory (resolve_runtime_dir's result — already the +// per-user default, /run/user//velox, unless XDG_RUNTIME_DIR says otherwise). A +// second daemon pointed at the same runtime dir gets EADDRINUSE and exits; one pointed at +// a different (isolated / test) runtime dir gets its own lock and starts fine. The kernel +// reclaims an abstract-namespace address when the holding process dies, so a crash never +// wedges it (docs/01 §2). +// +// Naming this "velox-daemon-" alone (the old scheme) meant exactly one name per +// user system-wide, so XDG_RUNTIME_DIR isolation never reached it: a leaked test veloxd +// with the same euid held the lock for every isolated instance too, real or test, until +// it was killed. Hashing the resolved runtime dir path instead keeps the real per-user +// daemon unique (its runtime dir is unique to it) while letting isolated instances that +// each point at their own runtime dir coexist. + +#include + +namespace velox::daemon::rpc { + +// Returns the held fd (kept open for the process lifetime; closing it releases the lock) +// or -1 if another process already holds the lock for this exact `runtime_dir`, or on any +// other socket error. +int acquire_single_instance_lock(const std::string& runtime_dir); + +} // namespace velox::daemon::rpc diff --git a/daemon/src/store/migrations/0003_start_mode_contract_values.sql b/daemon/src/store/migrations/0003_start_mode_contract_values.sql new file mode 100644 index 0000000..32cb6ef --- /dev/null +++ b/daemon/src/store/migrations/0003_start_mode_contract_values.sql @@ -0,0 +1,91 @@ +-- Migration 0003 — tasks.start_mode is rebuilt to the contract's StartMode values. +-- +-- 0001's CHECK read `start_mode IN ('auto','now','queue','manual')`. That is not +-- StartMode.schema.json's enum at all: the contract is ['now','later','queue']. The +-- practical effect: `download.add` with `startMode: "later"` — a real, documented value +-- (the File Info dialog's Download Later button) — hit the CHECK constraint on insert +-- and surfaced as an unhandled -32603, every time. 'auto' and 'manual' were never +-- contract values; they were this table's own invention and nothing on the wire ever +-- sends them. +-- +-- SQLite cannot ALTER a CHECK constraint in place, so this rebuilds the table (same +-- pattern as 0002: create the new shape, copy with the value mapped, drop, rename). +-- Existing rows are remapped by what they actually meant: 'auto' was "eligible for the +-- scheduler the moment it's added", i.e. 'now'; 'manual' was "parked, wait for the user", +-- which is what 'later' means on the wire (StartMode's own description: "lands the task +-- in paused"). Any row already spelled 'now' or 'queue' passes through unchanged. +-- +-- This does NOT touch `state` or `pause_reason` — a row that was start_mode='manual' and +-- (per the dispatcher's now-dead branch) state='new' keeps state='new'; the daemon-side +-- fix to actually land a 'later' task in 'paused' going forward lives in dispatcher.cpp, +-- not in this migration. Historical rows are not replayed through the scheduler. + +CREATE TABLE tasks_new ( + task_id TEXT PRIMARY KEY, + url TEXT NOT NULL, + effective_url TEXT, + filename TEXT NOT NULL DEFAULT '', + save_dir TEXT NOT NULL, + category_id TEXT REFERENCES categories(category_id) ON DELETE SET NULL, + queue_id TEXT REFERENCES queues(queue_id) ON DELETE SET NULL, + queue_position INTEGER, + + state TEXT NOT NULL DEFAULT 'new' + CHECK (state IN ('new','probing','queued','connecting','downloading','paused', + 'retry_wait','assembling','verifying','complete','failed','cancelled')), + pause_reason TEXT CHECK (pause_reason IN + ('user','schedule','queue_stopped','admission_reconcile','auto')), + + size_bytes INTEGER, + downloaded_bytes INTEGER NOT NULL DEFAULT 0, + resumable INTEGER NOT NULL DEFAULT 0, + + req_segments INTEGER, + eff_segments INTEGER NOT NULL DEFAULT 0, + req_buffer_bytes INTEGER, + eff_buffer_bytes INTEGER, + + -- The contract's StartMode (StartMode.schema.json): 'now' | 'later' | 'queue'. + start_mode TEXT NOT NULL DEFAULT 'now' + CHECK (start_mode IN ('now','later','queue')), + description TEXT, + + etag TEXT, + last_modified TEXT, + content_type TEXT, + + checksum_algo TEXT CHECK (checksum_algo IN ('md5','sha1','sha256','sha512')), + checksum_value TEXT, + + error_code TEXT, + error_message TEXT, + error_http_status INTEGER, + error_retryable INTEGER, + error_attempt INTEGER, + error_next_retry_at TEXT, + + speed_bps INTEGER NOT NULL DEFAULT 0, + + created_at TEXT NOT NULL, + last_try_at TEXT, + completed_at TEXT +) STRICT; + +INSERT INTO tasks_new + SELECT task_id, url, effective_url, filename, save_dir, category_id, queue_id, + queue_position, state, pause_reason, size_bytes, downloaded_bytes, resumable, + req_segments, eff_segments, req_buffer_bytes, eff_buffer_bytes, + CASE start_mode WHEN 'auto' THEN 'now' WHEN 'manual' THEN 'later' ELSE start_mode END, + description, etag, last_modified, content_type, checksum_algo, checksum_value, + error_code, error_message, error_http_status, error_retryable, error_attempt, + error_next_retry_at, speed_bps, created_at, last_try_at, completed_at + FROM tasks; + +DROP TABLE tasks; +ALTER TABLE tasks_new RENAME TO tasks; + +CREATE INDEX idx_tasks_state ON tasks(state); +CREATE INDEX idx_tasks_category ON tasks(category_id); +CREATE INDEX idx_tasks_queue_order ON tasks(queue_id, queue_position); +CREATE INDEX idx_tasks_created ON tasks(created_at); +CREATE INDEX idx_tasks_completed ON tasks(completed_at); diff --git a/daemon/src/store/tasks.hpp b/daemon/src/store/tasks.hpp index 71b33db..4012a7b 100644 --- a/daemon/src/store/tasks.hpp +++ b/daemon/src/store/tasks.hpp @@ -21,7 +21,7 @@ struct TaskRow { std::string save_dir; std::string filename; std::string state = "new"; - std::string start_mode = "auto"; + std::string start_mode = "now"; // contract StartMode: 'now'|'later'|'queue' std::string created_at; std::optional effective_url; diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index 9c84ca4..91803c6 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -23,3 +23,4 @@ veloxd_test(store_tasks LIBS veloxd_store) veloxd_test(sched_scheduler LIBS veloxd_sched veloxd_rpc) veloxd_test(event_hub LIBS veloxd_rpc) veloxd_test(store_categories_queues LIBS veloxd_store) +veloxd_test(single_instance LIBS veloxd_rpc) diff --git a/daemon/tests/single_instance_test.cpp b/daemon/tests/single_instance_test.cpp new file mode 100644 index 0000000..60e5e2d --- /dev/null +++ b/daemon/tests/single_instance_test.cpp @@ -0,0 +1,40 @@ +// The single-instance lock: isolated by runtime dir, not just by euid. This is the bug a +// leaked test veloxd exploited — one abstract-socket name per user meant every isolated +// instance (real daemon, tests, other lanes) fought over the same lock. + +#include + +#include "check.hpp" +#include "rpc/single_instance.hpp" + +using namespace velox::daemon::rpc; + +void run() { + // Two different runtime dirs: both acquire the lock independently. + { + const int a = acquire_single_instance_lock("/run/user/1000/velox-test-a"); + const int b = acquire_single_instance_lock("/run/user/1000/velox-test-b"); + CHECK(a >= 0); + CHECK(b >= 0); + if (a >= 0) ::close(a); + if (b >= 0) ::close(b); + } + + // Same runtime dir: the second attempt is refused while the first still holds it. + { + const int first = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + CHECK(first >= 0); + const int second = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + CHECK(second < 0); + if (first >= 0) ::close(first); + if (second >= 0) ::close(second); + + // Releasing (closing) the fd frees the abstract-namespace name immediately — a + // third attempt at the same dir succeeds once the first is gone. + const int third = acquire_single_instance_lock("/run/user/1000/velox-test-shared"); + CHECK(third >= 0); + if (third >= 0) ::close(third); + } +} + +TEST_MAIN() diff --git a/daemon/tests/store_migrations_test.cpp b/daemon/tests/store_migrations_test.cpp index 76e97d9..9b41389 100644 --- a/daemon/tests/store_migrations_test.cpp +++ b/daemon/tests/store_migrations_test.cpp @@ -86,6 +86,56 @@ void run() { .has_value()); } + // --- 0003: start_mode is rebuilt to the contract's values, existing rows mapped - + { + auto db = Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + // Build a real pre-0003 db (schema 1..2) with rows in the old, non-contract + // start_mode spelling, the way an actually-released daemon would have them. + for (const auto& m : embedded_migrations()) { + if (m.version > 2) break; + CHECK(db->exec(m.sql).has_value()); + CHECK(db->set_user_version(m.version).has_value()); + } + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('auto1','http://x','/tmp','2026-09-10T00:00:00Z','auto')") + .has_value()); + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('man1','http://x','/tmp','2026-09-10T00:00:00Z','manual')") + .has_value()); + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('q1','http://x','/tmp','2026-09-10T00:00:00Z','queue')") + .has_value()); + + CHECK(migrate_to_head(*db).has_value()); + CHECK_EQ(db->user_version(), head); + + auto start_mode_of = [&](const char* id) -> std::string { + auto st = db->prepare("SELECT start_mode FROM tasks WHERE task_id=?1"); + if (!st || !st->bind(1, std::string_view(id))) return ""; + auto row = st->step(); + if (!row || !*row) return ""; + return std::string(st->column_text(0)); + }; + CHECK_EQ(start_mode_of("auto1"), std::string("now")); + CHECK_EQ(start_mode_of("man1"), std::string("later")); + CHECK_EQ(start_mode_of("q1"), std::string("queue")); // passes through unchanged + + // The bug this migration closes: 'later' — a real, documented StartMode value — + // used to hit the old CHECK and fail every insert. It's accepted now, and the two + // retired spellings are gone for good. + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('later1','http://x','/tmp','2026-09-10T00:00:00Z','later')") + .has_value()); + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('bad1','http://x','/tmp','2026-09-10T00:00:00Z','auto')") + .has_value() == false); + CHECK(db->exec("INSERT INTO tasks(task_id,url,save_dir,created_at,start_mode) " + "VALUES('bad2','http://x','/tmp','2026-09-10T00:00:00Z','manual')") + .has_value() == false); + } + // --- re-running the migrator on an at-head DB is a no-op ------------------------ { auto db = Db::open(":memory:"); From 55f0c6099de52c3c6019ff07ade929bf61d22237 Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:20:58 +0400 Subject: [PATCH 2/5] =?UTF-8?q?daemon:=20D4b=20=E2=80=94=20download.pause/?= =?UTF-8?q?resume/start/cancel=20and=20queue.start/stop=20drive=20the=20sc?= =?UTF-8?q?heduler?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit download.pause/resume/start/cancel and queue.start/stop were stubs; now they call into the scheduler and take effect immediately, not on the next 1s tick — pausing, resuming or cancelling a live transfer can't wait, 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 what dispatcher.hpp depends on instead of sched/scheduler.hpp directly. Needed because veloxd_sched already links veloxd_rpc (for EventHub); dispatcher.hpp pulling in sched/scheduler.hpp directly would make it a real circular library dependency, breaking anything that links veloxd_rpc alone (cli's tests, as it turned out — hit and fixed during this change). Scheduler::user_pause/user_resume/user_start/user_cancel + pause_queue follow tick()'s existing to_pause pattern: call the engine (async, no synchronous effect) and transition the store eagerly so download.get/list are correct the instant the RPC call returns. Fixed a real bug surfaced while building this: transition() always overwrote pause_reason to NULL when the engine's own delayed pause-ack callback (on_state to paused, no error) arrived after whoever actually initiated the pause had already written the real reason — now it preserves the stored reason when the callback supplies none, instead of clobbering it. Covered by a regression check in sched_scheduler_test. store/queues gets get() and set_state() (was list()-only) for queue.start/stop. Verified against real veloxd + tools/testserver, not just unit tests: pausing a live single-segment throttled transfer freezes downloadedBytes, resume continues it from that point, cancel stops it; a bad taskId comes back in BulkTaskResult.failed with -32010, not a crash; queue.stop(pauseRunning:true) pauses the queue's running task immediately and queue.start resumes admission. Known gap: download.start's contract "a task in 'queued' jumps its queue" (priority bump) is not implemented — admission is still plain FIFO by created_at. Noted in deferrals.md. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/CMakeLists.txt | 10 ++- daemon/docs/deferrals.md | 6 +- daemon/src/main.cpp | 2 +- daemon/src/rpc/dispatcher.cpp | 106 +++++++++++++++++++++++--- daemon/src/rpc/dispatcher.hpp | 12 ++- daemon/src/rpc/task_action_port.hpp | 44 +++++++++++ daemon/src/sched/scheduler.cpp | 105 ++++++++++++++++++++++++- daemon/src/sched/scheduler.hpp | 44 ++++++++++- daemon/src/store/queues.cpp | 83 ++++++++++++++------ daemon/src/store/queues.hpp | 9 +++ daemon/tests/sched_scheduler_test.cpp | 102 +++++++++++++++++++++++++ 11 files changed, 478 insertions(+), 45 deletions(-) create mode 100644 daemon/src/rpc/task_action_port.hpp 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() From 4f6fb1029df6b7cf85ef73ea6115c623c94fa1ba Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:30:41 +0400 Subject: [PATCH 3/5] =?UTF-8?q?daemon:=20D2=20=E2=80=94=20download.probe,?= =?UTF-8?q?=20genuinely=20async=20on=20both=20transports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit download.probe was a stub (-32603). It needs an HTTP round trip on the engine's probe pool (up to the schema's 30s x-deadlineMs), which cannot fit VeloxDispatcher's synchronous on_download_probe -> HandlerResult return without blocking the RPC loop for the duration — a hard no per CLAUDE.md ("never block the RPC loop") and AGENT-DAEMON.md build step 1. uds_server.cpp and ws_server.cpp special-case "download.probe" before the generic dispatch(), exactly the way they already special-case session.hello/session.subscribe: parse the params, call the port, and queue the reply whenever the callback fires (dropped silently if the connection is gone by then). rpc::TaskActionPort gains probe_now(DownloadProbeParams, callback) — kept in proto/std terms, no vdm::net::* in the signature, so veloxd_rpc never needs core/include's vdm headers just to declare this. sched::Scheduler::probe_now is the implementation: builds a vdm::net::ProbeRequest, runs it on the engine's probe pool, marshals the engine-thread callback back onto the loop (deps_.post_to_loop, same as every other engine callback here), maps a probe failure to -32013 ProbeFailed (data.httpStatus set when there was an HTTP response), and fills suggestedCategoryId/suggestedSaveDir with a plain extension match against the categories table — not the real rules engine, which is still D3; noted in a comment. Verified against real veloxd + tools/testserver, not just unit tests: a real probe answers in ~5ms with size/resumable/etag/redirect chain; a bad host maps to -32013; and — the actual point of the async design — a connection running a 10s slow-loris probe does not block a second connection's download.list, which answers in ~1ms while the probe is still outstanding. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 2 +- daemon/src/main.cpp | 4 +- daemon/src/rpc/task_action_port.hpp | 19 +++++++ daemon/src/rpc/uds_server.cpp | 40 +++++++++++++- daemon/src/rpc/uds_server.hpp | 14 ++++- daemon/src/rpc/ws_server.cpp | 39 ++++++++++++- daemon/src/rpc/ws_server.hpp | 12 +++- daemon/src/sched/scheduler.cpp | 80 ++++++++++++++++++++++++++- daemon/src/sched/scheduler.hpp | 18 +++--- daemon/tests/sched_scheduler_test.cpp | 45 +++++++++++++++ 10 files changed, 255 insertions(+), 18 deletions(-) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 52aad48..3d3d8e7 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -6,7 +6,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | 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.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`) | +| ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | | 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~~ | **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 | diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index 788ce4a..7b1160d 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -161,7 +161,7 @@ int main() { }); } - velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path()); + velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path(), &scheduler); if (const auto ec = uds.start()) { std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message() << "\n"; @@ -175,7 +175,7 @@ int main() { // TODO(build step 7): replace EnvAutoApprover with a GUI-dialog / desktop-notification // approver. Until then pairing needs VELOX_PAIR_AUTO=1. velox::daemon::rpc::EnvAutoApprover approver; - velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt); + velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt, &scheduler); if (const auto ec = ws.start()) { std::cerr << "veloxd: WebSocket transport unavailable (" << ec.message() << "); the extension fallback will not work this run\n"; diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index 6f03218..ce39e67 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -12,9 +12,12 @@ // 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 #include +#include "velox_proto.hpp" + namespace velox::daemon::rpc { class TaskActionPort { @@ -39,6 +42,22 @@ public: // 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; + + // download.probe (D2), the File Info dialog's own network round trip — no task row + // involved. Genuinely async (the engine's probe pool; up to the schema's 30s + // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe: + // the RPC server layer (uds_server.cpp / ws_server.cpp) special-cases "download.probe" + // before the generic dispatch(), the same way it already special-cases session.hello, + // calls this, and queues the reply whenever `done` fires — on an engine thread, so the + // implementation must marshal back to the loop before calling it, the same as every + // other EnginePort callback. Kept in std::string/proto terms (not vdm::net::*) so this + // header — included by dispatcher.hpp, part of veloxd_rpc — never needs core/include's + // vdm headers; the vdm::net::ProbeRequest/ProbeResult conversion lives in sched/, which + // already depends on vdm. + virtual void probe_now( + const velox::proto::DownloadProbeParams& params, + std::function)> + done) = 0; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp index 5b3c36c..aaf0ac4 100644 --- a/daemon/src/rpc/uds_server.cpp +++ b/daemon/src/rpc/uds_server.cpp @@ -57,8 +57,9 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json } // namespace UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, EventHub& hub, - std::string socket_path) - : loop_(loop), dispatcher_(dispatcher), hub_(hub), path_(std::move(socket_path)) {} + std::string socket_path, TaskActionPort* actions) + : loop_(loop), dispatcher_(dispatcher), hub_(hub), actions_(actions), + path_(std::move(socket_path)) {} UdsServer::~UdsServer() { for (auto& [fd, c] : conns_) { @@ -202,12 +203,47 @@ void UdsServer::handle_line(Conn& c, const std::string& line) { } } + if (method == "download.probe") { + handle_download_probe(c, req); + return; + } + // Everything else: the generated router. It returns a null json for a notification // that needs no reply. json reply = proto::dispatch(dispatcher_, proto::Transport::Uds, req); if (!reply.is_null()) queue_reply(c, reply); } +void UdsServer::handle_download_probe(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.probe")); + return; + } + + const int fd = c.fd; + actions_->probe_now( + *parsed, [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + queue_reply(*it->second, proto::make_result(id, *r)); + } else { + queue_reply(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool UdsServer::handle_session_method(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/uds_server.hpp b/daemon/src/rpc/uds_server.hpp index dd88faf..d76f7a8 100644 --- a/daemon/src/rpc/uds_server.hpp +++ b/daemon/src/rpc/uds_server.hpp @@ -22,6 +22,7 @@ #include "rpc/event_hub.hpp" #include "rpc/ndjson.hpp" +#include "rpc/task_action_port.hpp" #include "velox_proto.hpp" namespace velox::daemon::rpc { @@ -30,8 +31,11 @@ class EventLoop; class UdsServer { public: + // `actions` is optional (nullptr in tests that don't need download.probe) — see + // handle_download_probe's own comment for why this method can't go through the + // generic dispatch() path like everything else. UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, EventHub& hub, - std::string socket_path); + std::string socket_path, TaskActionPort* actions = nullptr); ~UdsServer(); UdsServer(const UdsServer&) = delete; @@ -66,6 +70,13 @@ private: bool handle_session_method(Conn& c, const std::string& method, const nlohmann::json& request, nlohmann::json& reply); + // download.probe is genuinely async (up to the schema's 30s x-deadlineMs, on the + // engine's probe pool) and so cannot fit the synchronous generic dispatch() path — + // special-cased here exactly the way handle_session_method special-cases session.*. + // Queues the reply itself, later, when actions_->probe_now()'s callback fires; does + // nothing if the connection is gone by then (client disconnected mid-probe). + void handle_download_probe(Conn& c, const nlohmann::json& request); + void queue_reply(Conn& c, const nlohmann::json& reply); void flush(Conn& c); void close_conn(int fd); @@ -73,6 +84,7 @@ private: EventLoop& loop_; velox::proto::Dispatcher& dispatcher_; EventHub& hub_; + TaskActionPort* actions_; std::string path_; int listen_fd_ = -1; bool bound_ = false; // path_ is ours to unlink on destruction diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp index a3f2aa3..875bee5 100644 --- a/daemon/src/rpc/ws_server.cpp +++ b/daemon/src/rpc/ws_server.cpp @@ -60,12 +60,14 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json } // namespace WsServer::WsServer(EventLoop& loop, proto::Dispatcher& dispatcher, store::Db& db, - PairingApprover& approver, EventHub& hub, RuntimeDir runtime) + PairingApprover& approver, EventHub& hub, RuntimeDir runtime, + TaskActionPort* actions) : loop_(loop), dispatcher_(dispatcher), db_(db), approver_(approver), hub_(hub), + actions_(actions), runtime_(std::move(runtime)) {} WsServer::~WsServer() { @@ -261,10 +263,45 @@ void WsServer::handle_rpc(Conn& c, const std::string& text) { return; } + if (method == "download.probe") { + handle_download_probe(c, req); + return; + } + json reply = proto::dispatch(dispatcher_, proto::Transport::Ws, req); if (!reply.is_null()) send_text(c, reply); } +void WsServer::handle_download_probe(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + send_text(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + send_text(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.probe")); + return; + } + + const int fd = c.fd; + actions_->probe_now( + *parsed, [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + send_text(*it->second, proto::make_result(id, *r)); + } else { + send_text(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/ws_server.hpp b/daemon/src/rpc/ws_server.hpp index 2f8e8a9..bc71864 100644 --- a/daemon/src/rpc/ws_server.hpp +++ b/daemon/src/rpc/ws_server.hpp @@ -18,6 +18,7 @@ #include "rpc/event_hub.hpp" #include "rpc/pairing.hpp" #include "rpc/runtime_dir.hpp" +#include "rpc/task_action_port.hpp" #include "rpc/ws_frame.hpp" #include "velox_proto.hpp" @@ -31,8 +32,12 @@ class EventLoop; class WsServer { public: + // `actions` is optional (nullptr in tests that don't need download.probe) — see + // handle_download_probe's own comment for why this method can't go through the + // generic dispatch() path like everything else. WsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, velox::daemon::store::Db& db, - PairingApprover& approver, EventHub& hub, RuntimeDir runtime); + PairingApprover& approver, EventHub& hub, RuntimeDir runtime, + TaskActionPort* actions = nullptr); ~WsServer(); WsServer(const WsServer&) = delete; @@ -73,6 +78,10 @@ private: bool handle_session_ws(Conn& c, const std::string& method, const nlohmann::json& request, nlohmann::json& reply); + // See UdsServer::handle_download_probe — same reasoning, same pattern, duplicated per + // transport because each owns its own Conn/send mechanics. + void handle_download_probe(Conn& c, const nlohmann::json& request); + void send_text(Conn& c, const nlohmann::json& value); void send_frame(Conn& c, WsOpcode op, std::string_view payload); void begin_close(Conn& c, std::uint16_t code, std::string_view reason); @@ -84,6 +93,7 @@ private: velox::daemon::store::Db& db_; PairingApprover& approver_; EventHub& hub_; + TaskActionPort* actions_; RuntimeDir runtime_; PairingRateLimiter rate_limiter_; diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index f460a53..ac296b5 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -6,6 +6,7 @@ #include #include "sched/schedule_window.hpp" +#include "store/categories.hpp" #include "store/segments.hpp" #include "store/settings.hpp" #include "store/tasks.hpp" @@ -570,9 +571,82 @@ std::vector Scheduler::pause_queue(const std::string& queue_id) { return paused; } -void Scheduler::probe_now(const vdm::net::ProbeRequest& req, - std::function)> done) { - engine_.probe(req, std::move(done)); +namespace { + +// Extension match against the categories table (categories.extensions, per category.list), +// the same table download.add would consult once rules.* actually exists (D3). Not the +// real rules engine — no host/mime/size clauses — just enough that the File Info dialog's +// preselect isn't always "general". +std::string guess_category_id(store::Db& db, const std::string& filename) { + const auto dot = filename.find_last_of('.'); + if (dot == std::string::npos || dot + 1 >= filename.size()) return "general"; + std::string ext = filename.substr(dot + 1); + std::transform(ext.begin(), ext.end(), ext.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + + store::Categories categories(db); + auto cats = categories.list(); + if (!cats) return "general"; + for (const auto& c : *cats) { + for (const auto& e : c.extensions) + if (e == ext) return c.categoryId; + } + return "general"; +} + +} // namespace + +void Scheduler::probe_now( + const proto::DownloadProbeParams& params, + std::function)> done) { + vdm::net::ProbeRequest req; + req.url = params.url; + if (params.headers) + for (const auto& [k, v] : *params.headers) req.headers.push_back({k, v}); + if (params.cookies) + for (const auto& c : *params.cookies) req.cookies.push_back({c.name, c.value}); + if (params.referrer) req.referrer = *params.referrer; + if (params.userAgent) req.user_agent = *params.userAgent; + + engine_.probe(req, [this, params, done](vdm::Result pr) { + deps_.post_to_loop([this, params, done, pr]() { + if (!pr) { + nlohmann::json data; + if (pr.error().http_status != 0) data["httpStatus"] = pr.error().http_status; + done(std::unexpected(proto::HandlerError{ + proto::ErrorCode::ProbeFailed, pr.error().context, data})); + return; + } + + const std::string filename = vdm::net::suggest_filename(*pr); + + proto::DownloadProbeResult r; + r.filename = filename.empty() ? "download.bin" : filename; + if (pr->total_size) r.sizeBytes = static_cast(*pr->total_size); + r.mime = pr->mime; + r.resumable = pr->resumable; + r.effectiveUrl = pr->effective_url.empty() ? params.url : pr->effective_url; + r.suggestedCategoryId = guess_category_id(db_, r.filename); + if (!pr->etag.empty()) r.etag = pr->etag; + if (!pr->last_modified.empty()) r.lastModified = pr->last_modified; + r.acceptRanges = pr->accept_ranges; + if (!pr->redirect_chain.empty()) r.redirectChain = pr->redirect_chain; + if (pr->requires_auth) r.requiresAuth = true; + + store::Settings settings(db_); + store::Categories categories(db_); + if (auto cats = categories.list()) { + for (const auto& c : *cats) + if (c.categoryId == r.suggestedCategoryId) { + r.suggestedSaveDir = c.saveDir; + break; + } + } + if (!r.suggestedSaveDir) r.suggestedSaveDir = settings.get_string("saveTo.defaultDir"); + + done(r); + }); + }); } } // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index d9db9ae..29aba9a 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -132,13 +132,17 @@ public: // 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); + // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the + // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the + // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a + // plain extension match against the categories table; the real rules engine is D3). + // `done` is called already marshalled onto the loop thread via post_to_loop, same as + // every other engine callback here — the caller never has to know it started on an + // engine thread. + void probe_now( + const velox::proto::DownloadProbeParams& params, + std::function)> + done) override; // Diagnostics / tests. std::optional wire_id_of(vdm::TaskId id) const; diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp index 11f281c..8f90977 100644 --- a/daemon/tests/sched_scheduler_test.cpp +++ b/daemon/tests/sched_scheduler_test.cpp @@ -361,6 +361,51 @@ void run() { tasks.remove("uc1"); CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value()); } + + // --- probe_now: success (with a category guess) and a mapped failure ------------ + { + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + + vdm::net::ProbeResult pr; + pr.effective_url = "https://cdn.example/movie.mp4"; + pr.filename_from_url = "movie.mp4"; // suggest_filename() needs this set; the real + // Prober fills it from the URL path itself + pr.total_size = 123456; + pr.mime = "video/mp4"; + pr.resumable = true; + pr.accept_ranges = true; + pr.etag = "\"abc\""; + engine.auto_probe_result = vdm::Result{pr}; + + velox::proto::DownloadProbeParams params; + params.url = "https://cdn.example/movie.mp4"; + std::optional> got; + sched.probe_now(params, [&](auto r) { got = std::move(r); }); + CHECK(got.has_value()); + CHECK(got->has_value()); + if (got && *got) { + const auto& r = **got; + CHECK_EQ(r.sizeBytes.value_or(-1), std::int64_t{123456}); + CHECK(r.resumable); + CHECK_EQ(r.mime, std::string("video/mp4")); + // movie.mp4 -> the 'video' built-in category by extension. + CHECK_EQ(r.suggestedCategoryId, std::string("video")); + } + + vdm::ErrorInfo err; + err.code = vdm::Error::connect_failed; + err.context = "connection refused"; + engine.auto_probe_result = vdm::Result{err}; + std::optional> got2; + sched.probe_now(params, [&](auto r) { got2 = std::move(r); }); + CHECK(got2.has_value()); + CHECK(!got2->has_value()); + if (got2 && !*got2) + CHECK(got2->error().code == velox::proto::ErrorCode::ProbeFailed); + } } TEST_MAIN() From c89158ea09cd375733d0fbe0b0664a239dcf1c80 Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:43:43 +0400 Subject: [PATCH 4/5] =?UTF-8?q?daemon:=20D3=20(partial)=20=E2=80=94=20cate?= =?UTF-8?q?gory.upsert/remove,=20queue.upsert,=20download.remove/addBatch/?= =?UTF-8?q?provideAuth?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes a bounded, high-value slice of the remaining D3 stubs. settings.*/rules.*/ limiter.*/schedule.*/grabber.*/media.*/capture.*/queue.reorder/download.refreshUrl/ download.update stay deferred — reasons noted individually in deferrals.md (settings.* specifically: a real, large field<->key<->JSON-type mapping table across ~43 fields / ~50 SettingKeys, not started rather than rushed). category.upsert/remove: store/categories.hpp gains get/upsert/remove. upsert generates an id when absent and always ignores the payload's `builtin` (preserved from the existing row on replace, false on create); saveDir goes through the same fs::resolve_target canonicalize-and-root-check as download.add. remove refuses a builtin at both layers (dispatcher's -32602 pre-check; the store's own "DELETE ... AND builtin = 0" as defense in depth) and reassigns member tasks to reassignTo (default "general") inside one transaction before deleting the row. queue.upsert: store/queues.hpp gains get/upsert (set_state already existed from D4b). Same create-generates-id pattern; a create always starts 'stopped', a replace keeps the queue's current run state (upsert edits config, not run state — that's queue.start/stop). Also fixed in passing: on_complete has been a real column since migration 0001 but Queues::list/get never projected it onto Queue.onComplete. download.remove: cancels with discard_partial=true (always drops the .veloxpart pair — unlike download.cancel, which keeps them, the row is gone either way), deletes the finished file only when deleteFile is true and the task was complete (best-effort), deletes the row (segments cascade via the FK), and publishes event.task.removed (closing the last open note under D5). download.addBatch: on_download_add's body is now a shared add_one(), called once per item after merging each item's unset fields against params.defaults. download.provideAuth: forwards to EnginePort::provide_auth via a new TaskActionPort::provide_auth. `remember`/persisting to the Secret Service is accepted but not acted on — nothing in this build talks to libsecret yet. Verified against real veloxd + tools/testserver: category create/replace/ remove-with-reassignment, queue create/replace-keeps-run-state, a batch add sharing defaults.saveDir, and download.remove with deleteFile actually deleting the file and the task then 404ing download.get with -32010. Full ctest: 39/39. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 6 +- daemon/src/rpc/dispatcher.cpp | 163 ++++++++++++++++-- daemon/src/rpc/dispatcher.hpp | 6 + daemon/src/rpc/task_action_port.hpp | 8 + daemon/src/sched/scheduler.cpp | 9 + daemon/src/sched/scheduler.hpp | 3 + daemon/src/store/categories.cpp | 139 +++++++++++++-- daemon/src/store/categories.hpp | 24 +++ daemon/src/store/queues.cpp | 59 ++++++- daemon/src/store/queues.hpp | 6 + daemon/tests/store_categories_queues_test.cpp | 101 +++++++++++ 11 files changed, 493 insertions(+), 31 deletions(-) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 3d3d8e7..588553a 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -7,7 +7,11 @@ 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~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | -| 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 | +| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet, or (`settings.*`) sound but large — see the note below. `category.list/upsert/remove`, `queue.list/upsert/start/stop`, `download.remove/addBatch/provideAuth` are done | Per method, as each wires to the store/scheduler | +| — | **`settings.get`/`settings.set` specifically, not started:** `proto::Settings` is a flat struct of ~43 `std::optional` fields, one per `SettingKey` (~50 keys) in `Settings.schema.json`; `store::Settings` already has `get_raw`/`set_raw`/`overrides` keyed by the same dotted strings the JSON uses (`"connection.maxSegmentsPerDownload"`, …). The handlers are a mechanical field <-> key <-> JSON-type mapping table in both directions (get: row-or-default -> struct field; set: struct field -> validate against the key's schema type -> `set_raw`, collecting `changed`) — real work, just long and repetitive rather than hard. Left alone this pass rather than rushed; every other read of settings in this codebase already goes through `store::Settings`'s typed helpers directly (`reload_config`, `on_download_add`'s segment default, `capture.offer`'s allowed roots when that lands), so nothing downstream is blocked on the RPC surface existing. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}` | the mapping table is genuinely large, not genuinely hard | its own pass | +| ~~D3a~~ | **Closed** — `category.upsert`/`category.remove`: `store/categories.hpp` gains `get`/`upsert`/`remove`. `upsert` generates an id when absent (create) and always ignores the payload's `builtin` (preserved from the existing row on replace, false on create — a client can never mint or revoke it); the `saveDir` goes through the same `fs::resolve_target` canonicalize-and-root-check as `download.add` (`-32011` on failure). `remove` refuses a builtin at both layers (dispatcher pre-checks for the `-32602` error text; the store's own `DELETE ... AND builtin = 0` is defense in depth) and reassigns member tasks to `reassignTo` (default `"general"`) inside one transaction before deleting the row. Note: the `categories` table (0001) has no columns for `Category.mimeTypes`/`.sortOrder` — accepted on `upsert` but not persisted. | `store/categories.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done, `mimeTypes`/`sortOrder` gap noted | +| ~~D3b~~ | **Closed** — `queue.upsert`: `store/queues.hpp` gains `get`/`upsert` (`set_state` already existed from D4b). Same create-generates-id pattern as categories; `taskIds` in the payload is ignored (schema's own note) and a create always starts `'stopped'` while a replace keeps the queue's current run state — `queue.upsert` edits config, not run state (that's `queue.start`/`stop`). Also fixed: `on_complete` was a real column since 0001 but `Queues::list`/`get` never projected it onto `Queue.onComplete` — now they do. | `store/queues.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done | +| ~~D3c~~ | **Closed** — `download.remove`: cancels with `discard_partial=true` through `TaskActionPort` (always drops any `.veloxpart`/`.veloxpart.meta` — the row is gone either way, unlike `download.cancel`, which keeps them), deletes the finished file only when `deleteFile` is true and the task was `complete` (best-effort — a missing file doesn't fail the call), deletes the row (segments cascade via the FK), and publishes `event.task.removed` (closing the last open note under D5). `download.addBatch`: `on_download_add`'s body is now a shared `add_one()`, called once per item after merging each item's unset fields against `params.defaults`. `download.provideAuth`: forwards to `EnginePort::provide_auth` through a new `TaskActionPort::provide_auth`; `remember`/persisting to the Secret Service is accepted but not acted on — nothing in this build talks to libsecret yet (verified: no such integration exists anywhere in the tree). Verified against real `veloxd` + `tools/testserver`: category create/replace/remove-with-reassignment, queue create/replace-keeps-state, a batch add with shared `defaults.saveDir`, and remove-with-deleteFile actually deleting the file and the task then 404ing `download.get` with `-32010`. | `rpc/dispatcher.{hpp,cpp}`, `rpc/task_action_port.hpp`, `sched/scheduler.{cpp,hpp}` | `download.provideAuth`'s `remember` (needs the Secret Service, unbuilt) | done, `remember` persistence gap noted | | ~~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~~ | **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 | diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 9f11f5d..c88d8be 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -1,5 +1,6 @@ #include "rpc/dispatcher.hpp" +#include #include #include @@ -141,9 +142,13 @@ VeloxDispatcher::on_download_list(const proto::DownloadListParams& params) { } // --- download.add : canonicalise + root-check the destination, then persist ----------- +// +// add_one() is the whole of download.add's body; on_download_add and on_download_addBatch +// (each item merged against `defaults` first) both call it, so there is exactly one place +// that turns a DownloadSpec into a stored, admitted task. -proto::HandlerResult -VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { +proto::HandlerResult VeloxDispatcher::add_one( + const proto::DownloadSpec& spec) { store::Settings settings(db_); std::string save_dir = spec.saveDir && !spec.saveDir->empty() @@ -226,6 +231,11 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { return r; } +proto::HandlerResult +VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { + return add_one(spec); +} + // --- everything else : not implemented until the store and scheduler land ------------- proto::HandlerResult @@ -248,16 +258,94 @@ VeloxDispatcher::on_category_list(const proto::CategoryListParams&) { return r; } proto::HandlerResult -VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) { - return not_implemented("category.remove"); +VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams& params) { + store::Categories categories(db_); + // A builtin category refuses with -32602 (category.remove's own words) — checked here + // rather than inferred from RemoveResult.removed=false, which also covers "no such + // category" and would otherwise collapse two different error stories into one. + auto existing = categories.get(params.categoryId); + if (!existing) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.remove: " + existing.error().message}); + if (!existing->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such category", + nlohmann::json{{"categoryId", params.categoryId}}}); + if ((*existing)->builtin) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "builtin categories cannot be removed", + nlohmann::json{{"categoryId", params.categoryId}}}); + + auto removed = categories.remove(params.categoryId, params.reassignTo); + if (!removed) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.remove: " + removed.error().message}); + proto::CategoryRemoveResult r; + r.removed = removed->removed; + r.reassignedTaskIds = removed->reassigned_task_ids; + if (on_mutation_) on_mutation_(); // download.list's categoryId column just moved + return r; } proto::HandlerResult -VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) { - return not_implemented("category.upsert"); +VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams& params) { + store::Settings settings(db_); + std::string save_dir = expand_tilde(params.category.saveDir); + + std::vector roots; + for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) { + if (auto c = fs::canonicalize_root(r)) roots.push_back(*c); + } + // No file is ever written for this marker leaf — resolve_target only validates/creates + // the directory chain (fs/safepath.hpp); it never creates the leaf itself. + auto target = fs::resolve_target(save_dir, ".category-marker", roots); + if (!target) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidPath, target.error().message, + nlohmann::json{{"path", params.category.saveDir}}}); + } + + store::Categories categories(db_); + proto::Category to_store = params.category; + to_store.saveDir = target->dir; + auto stored = categories.upsert(std::move(to_store)); + if (!stored) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "category.upsert: " + stored.error().message}); + proto::CategoryUpsertResult r; + r.category = std::move(*stored); + return r; } proto::HandlerResult -VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) { - return not_implemented("download.addBatch"); +VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams& params) { + proto::DownloadAddBatchResult r; + for (std::size_t i = 0; i < params.items.size(); ++i) { + proto::DownloadSpec spec = params.items[i]; // "its url is ignored" only for defaults + if (params.defaults) { + const auto& d = *params.defaults; + if (!spec.headers) spec.headers = d.headers; + if (!spec.cookies) spec.cookies = d.cookies; + if (!spec.referrer) spec.referrer = d.referrer; + if (!spec.userAgent) spec.userAgent = d.userAgent; + if (!spec.filename) spec.filename = d.filename; + if (!spec.saveDir) spec.saveDir = d.saveDir; + if (!spec.categoryId) spec.categoryId = d.categoryId; + if (!spec.queueId) spec.queueId = d.queueId; + if (!spec.segments) spec.segments = d.segments; + if (!spec.bufferBytes) spec.bufferBytes = d.bufferBytes; + if (!spec.startMode) spec.startMode = d.startMode; + if (!spec.description) spec.description = d.description; + if (!spec.checksum) spec.checksum = d.checksum; + } + + auto added = add_one(spec); + if (added) { + r.taskIds.push_back(added->taskId); + } else { + r.failed.push_back({static_cast(i), added.error().code, + added.error().message}); + } + } + return r; } proto::HandlerResult VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) { @@ -313,16 +401,56 @@ VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) { return not_implemented("download.probe"); } proto::HandlerResult -VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams&) { - return not_implemented("download.provideAuth"); +VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams& params) { + if (!actions_) return not_implemented("download.provideAuth"); + proto::DownloadProvideAuthResult r; + r.ok = actions_->provide_auth(params.taskId, params.username, params.password, + params.save.value_or(false)); + return r; } proto::HandlerResult VeloxDispatcher::on_download_refreshUrl(const proto::DownloadRefreshUrlParams&) { return not_implemented("download.refreshUrl"); } proto::HandlerResult -VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) { - return not_implemented("download.remove"); +VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams& params) { + store::Tasks tasks(db_); + proto::DownloadRemoveResult r; + for (const auto& id : params.taskIds) { + auto got = tasks.get(id); + if (!got || !got->has_value()) { + r.failed.push_back({id, proto::ErrorCode::TaskNotFound, "no such task"}); + continue; + } + const store::TaskRow row = **got; // copy: row is gone once tasks.remove() runs + + // Always discard the .veloxpart/.veloxpart.meta pair (the schema's own words) — + // download.remove deletes the row outright, so there is no reason to leave a + // partial behind the way download.cancel does. A no-op if the task never had a + // live engine handle (already terminal, or never started). + if (actions_) (void)actions_->user_cancel(id, /*discard_partial=*/true); + + // The finished file only when deleteFile is true — never touched for a partial or + // failed transfer (there is nothing there but what discard_partial above already + // took care of). Best-effort: a missing file is not a reason to fail the remove. + bool deleted_file = false; + if (params.deleteFile && row.state == "complete") { + std::error_code ec; + deleted_file = + std::filesystem::remove(std::filesystem::path(row.save_dir) / row.filename, ec); + } + + (void)tasks.remove(id); // segments cascade via the FK (ON DELETE CASCADE) + r.removed.push_back(id); + + hub_.publish(proto::Event::TaskRemoved, + proto::make_notification( + proto::Event::TaskRemoved, + nlohmann::json{{"taskId", id}, {"deletedFile", deleted_file}}), + id); + } + if (on_mutation_) on_mutation_(); + return r; } proto::HandlerResult VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) { @@ -436,8 +564,15 @@ VeloxDispatcher::on_queue_stop(const proto::QueueStopParams& params) { return r; } proto::HandlerResult -VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) { - return not_implemented("queue.upsert"); +VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams& params) { + store::Queues queues(db_); + auto stored = queues.upsert(params.queue); + if (!stored) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.upsert: " + stored.error().message}); + proto::QueueUpsertResult r; + r.queue = std::move(*stored); + return r; } proto::HandlerResult VeloxDispatcher::on_rules_list(const proto::RulesListParams&) { diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 2c4ccc5..89c1aad 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -116,6 +116,12 @@ public: on_settings_set(const velox::proto::SettingsSetParams&) override; private: + // The whole of download.add's body; on_download_add and on_download_addBatch (each + // item merged against DownloadAddBatchParams.defaults first) both call this — exactly + // one place turns a DownloadSpec into a stored, admitted task. + velox::proto::HandlerResult add_one( + const velox::proto::DownloadSpec& spec); + velox::daemon::store::Db& db_; EventHub& hub_; TaskActionPort* actions_; diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index ce39e67..a036e96 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -43,6 +43,14 @@ public: // Returns the wire ids actually paused. virtual std::vector pause_queue(const std::string& queue_id) = 0; + // download.provideAuth: answers a task auto-paused on a 401/407. false if the task + // isn't currently holding a live engine handle (nothing waiting on credentials). + // `remember` is accepted but not yet acted on — persisting to the Secret Service isn't + // wired anywhere in this build yet (CLAUDE.md §4: never SQLite, never logs); this + // always does the "this retry only" half. Noted in deferrals.md. + virtual bool provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) = 0; + // download.probe (D2), the File Info dialog's own network round trip — no task row // involved. Genuinely async (the engine's probe pool; up to the schema's 30s // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe: diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index ac296b5..61e128c 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -571,6 +571,15 @@ std::vector Scheduler::pause_queue(const std::string& queue_id) { return paused; } +bool Scheduler::provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) { + (void)remember; // not yet wired to the Secret Service anywhere in this build + auto eid = engine_id_of(wire_id); + if (!eid) return false; + engine_.provide_auth(*eid, username, password, remember); + return true; +} + namespace { // Extension match against the categories table (categories.extensions, per category.list), diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index 29aba9a..c020ac2 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -132,6 +132,9 @@ public: // instead of User. Returns the wire ids actually paused. std::vector pause_queue(const std::string& queue_id) override; + bool provide_auth(const std::string& wire_id, const std::string& username, + const std::string& password, bool remember) override; + // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a diff --git a/daemon/src/store/categories.cpp b/daemon/src/store/categories.cpp index 8532e8d..d585b4b 100644 --- a/daemon/src/store/categories.cpp +++ b/daemon/src/store/categories.cpp @@ -1,11 +1,35 @@ #include "store/categories.hpp" +#include + +#include +#include +#include + #include namespace velox::daemon::store { namespace proto = velox::proto; +namespace { + +proto::Category project_row(Stmt& st) { + proto::Category c; + c.categoryId = st.column_text(0); + c.name = st.column_text(1); + c.saveDir = st.column_text(2); + auto j = nlohmann::json::parse(st.column_text(3), nullptr, false); + if (j.is_array()) { + for (const auto& e : j) + if (e.is_string()) c.extensions.push_back(e.get()); + } + c.builtin = st.column_int(4) != 0; + return c; +} + +} // namespace + DbResult> Categories::list() { auto st = db_.prepare( "SELECT category_id, name, save_dir, extensions, builtin FROM categories " @@ -17,20 +41,111 @@ DbResult> Categories::list() { auto row = st->step(); if (!row) return std::unexpected(row.error()); if (!*row) break; - - proto::Category c; - c.categoryId = st->column_text(0); - c.name = st->column_text(1); - c.saveDir = st->column_text(2); - auto j = nlohmann::json::parse(st->column_text(3), nullptr, false); - if (j.is_array()) { - for (const auto& e : j) - if (e.is_string()) c.extensions.push_back(e.get()); - } - c.builtin = st->column_int(4) != 0; - out.push_back(std::move(c)); + out.push_back(project_row(*st)); } return out; } +DbResult> Categories::get(std::string_view category_id) { + auto st = db_.prepare( + "SELECT category_id, name, save_dir, extensions, builtin FROM categories " + "WHERE category_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, category_id); !b) return std::unexpected(b.error()); + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return std::optional{}; + return std::optional{project_row(*st)}; +} + +DbResult Categories::upsert(proto::Category category) { + // A replace keeps the existing row's builtin flag; a create is never builtin. Either + // way the payload's own `builtin` is ignored — a client cannot mint or revoke it. + bool builtin = false; + if (!category.categoryId.empty()) { + auto existing = get(category.categoryId); + if (!existing) return std::unexpected(existing.error()); + if (existing->has_value()) builtin = (*existing)->builtin; + } else { + // Reuse Tasks' id scheme (v4 UUID) would need a cross-module include for one + // function; a category id has no wire format requirement beyond "a string", so a + // timestamp-free random hex id keeps this module self-contained. + std::random_device rd; + std::uniform_int_distribution d; + char buf[17]; + std::snprintf(buf, sizeof(buf), "%016llx", static_cast(d(rd))); + category.categoryId = std::string(buf); + } + + nlohmann::json ext = nlohmann::json::array(); + for (const auto& e : category.extensions) ext.push_back(e); + + auto st = db_.prepare( + "INSERT INTO categories(category_id, name, save_dir, extensions, builtin) " + "VALUES(?1,?2,?3,?4,?5) " + "ON CONFLICT(category_id) DO UPDATE SET " + "name=excluded.name, save_dir=excluded.save_dir, extensions=excluded.extensions"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, std::string_view(category.categoryId)); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(2, std::string_view(category.name)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(3, std::string_view(category.saveDir)); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(4, std::string_view(ext.dump())); !b) return std::unexpected(b.error()); + if (auto b = st->bind(5, static_cast(builtin)); !b) + return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + + category.builtin = builtin; + return category; +} + +DbResult Categories::remove( + std::string_view category_id, const std::optional& reassign_to) { + RemoveResult out; + const std::string target = reassign_to.value_or("general"); + + // Db::transaction only threads a DbResult lambda; `out` is filled in-place and + // returned once the transaction (which may still fail and roll back) succeeds. + auto txn = db_.transaction([&]() -> DbResult { + { + // A builtin category is never removed, and — since it was never going to be + // removed — its tasks must not be reassigned away from it either. + auto chk = db_.prepare("SELECT builtin FROM categories WHERE category_id = ?1"); + if (!chk) return std::unexpected(chk.error()); + if (auto b = chk->bind(1, category_id); !b) return std::unexpected(b.error()); + auto row = chk->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) return {}; // no such category: removed stays false + if (chk->column_int(0) != 0) return {}; // builtin: removed stays false + } + { + auto sel = db_.prepare("SELECT task_id FROM tasks WHERE category_id = ?1"); + if (!sel) return std::unexpected(sel.error()); + if (auto b = sel->bind(1, category_id); !b) return std::unexpected(b.error()); + for (;;) { + auto row = sel->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + out.reassigned_task_ids.push_back(sel->column_text(0)); + } + } + if (!out.reassigned_task_ids.empty()) { + auto upd = db_.prepare("UPDATE tasks SET category_id = ?2 WHERE category_id = ?1"); + if (!upd) return std::unexpected(upd.error()); + if (auto b = upd->bind(1, category_id); !b) return std::unexpected(b.error()); + if (auto b = upd->bind(2, std::string_view(target)); !b) return std::unexpected(b.error()); + if (auto r = upd->step(); !r) return std::unexpected(r.error()); + } + auto del = db_.prepare("DELETE FROM categories WHERE category_id = ?1 AND builtin = 0"); + if (!del) return std::unexpected(del.error()); + if (auto b = del->bind(1, category_id); !b) return std::unexpected(b.error()); + if (auto r = del->step(); !r) return std::unexpected(r.error()); + out.removed = sqlite3_changes(db_.raw()) > 0; + return {}; + }); + if (!txn) return std::unexpected(txn.error()); + return out; +} + } // namespace velox::daemon::store diff --git a/daemon/src/store/categories.hpp b/daemon/src/store/categories.hpp index b4439e8..1c3b57c 100644 --- a/daemon/src/store/categories.hpp +++ b/daemon/src/store/categories.hpp @@ -3,7 +3,13 @@ // Read access to the `categories` table, projected onto proto::Category. Owned here // rather than duplicated per handler since category.list and download.add (rule // matching, later) both need it. +// +// The table has no columns for Category.mimeTypes / .sortOrder (0001_initial.sql predates +// those fields); upsert() accepts them but they are not persisted — round-tripped as unset +// on the next list()/get(). Noted in daemon/docs/deferrals.md. +#include +#include #include #include "store/sqlite.hpp" @@ -16,6 +22,24 @@ public: explicit Categories(Db& db) : db_(db) {} DbResult> list(); + DbResult> get(std::string_view category_id); + + // "Omit categoryId to create; supply it to replace" (category.upsert's own words) — + // the caller (dispatcher) decides create vs replace by whether `category.categoryId` + // is empty and generates the id; this just writes the row. `builtin` is never taken + // from the payload: preserved from the existing row on a replace, always false on a + // create (a client can never mint a builtin category). + DbResult upsert(velox::proto::Category category); + + // False for "no such category". A builtin category is never removed — the caller + // checks that (category.remove -> -32602) before calling this, since that check needs + // ErrorCode, which this module (like the rest of store/) does not depend on. + struct RemoveResult { + bool removed = false; + std::vector reassigned_task_ids; + }; + DbResult remove(std::string_view category_id, + const std::optional& reassign_to); private: Db& db_; diff --git a/daemon/src/store/queues.cpp b/daemon/src/store/queues.cpp index 9fb469f..8db8771 100644 --- a/daemon/src/store/queues.cpp +++ b/daemon/src/store/queues.cpp @@ -2,6 +2,10 @@ #include +#include +#include +#include + #include namespace velox::daemon::store { @@ -10,8 +14,8 @@ 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. +// One queue row (columns queue_id, name, state, max_concurrent, schedule, on_complete, 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); @@ -22,6 +26,7 @@ DbResult project_row(Db& db, Stmt& st) { auto j = nlohmann::json::parse(st.column_text(4), nullptr, false); if (auto sched = proto::parse(j, "schedule")) q.schedule = *sched; } + if (auto oc = proto::parse_QueueOnComplete(st.column_text(5))) q.onComplete = *oc; auto ts = db.prepare("SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position"); if (!ts) return std::unexpected(ts.error()); @@ -41,7 +46,8 @@ DbResult project_row(Db& db, Stmt& st) { DbResult> Queues::list() { auto st = db_.prepare( - "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name"); + "SELECT queue_id, name, state, max_concurrent, schedule, on_complete FROM queues " + "ORDER BY name"); if (!st) return std::unexpected(st.error()); std::vector out; @@ -58,7 +64,8 @@ DbResult> Queues::list() { 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"); + "SELECT queue_id, name, state, max_concurrent, schedule, on_complete 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(); @@ -78,4 +85,48 @@ DbResult Queues::set_state(std::string_view queue_id, std::string_view sta return sqlite3_changes(db_.raw()) > 0; } +DbResult Queues::upsert(proto::Queue queue) { + if (queue.queueId.empty()) { + std::random_device rd; + std::uniform_int_distribution d; + char buf[17]; + std::snprintf(buf, sizeof(buf), "%016llx", static_cast(d(rd))); + queue.queueId = std::string(buf); + } + + // A create defaults to 'stopped' (never auto-runs a brand-new queue); a replace keeps + // whatever run state the queue is already in — queue.upsert edits the config, not the + // run state (that's queue.start/stop). + std::string state = "stopped"; + if (auto existing = get(queue.queueId); existing && existing->has_value()) + state = std::string(proto::to_string((*existing)->state)); + + const std::string schedule_json = + queue.schedule ? nlohmann::json(*queue.schedule).dump() : std::string(); + const std::string on_complete = + std::string(proto::to_string(queue.onComplete.value_or(proto::QueueOnComplete::Nothing))); + + auto st = db_.prepare( + "INSERT INTO queues(queue_id, name, state, max_concurrent, schedule, on_complete) " + "VALUES(?1,?2,?3,?4,?5,?6) " + "ON CONFLICT(queue_id) DO UPDATE SET " + "name=excluded.name, max_concurrent=excluded.max_concurrent, " + "schedule=excluded.schedule, on_complete=excluded.on_complete"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, std::string_view(queue.queueId)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(2, std::string_view(queue.name)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(3, std::string_view(state)); !b) return std::unexpected(b.error()); + if (auto b = st->bind(4, queue.maxConcurrent); !b) return std::unexpected(b.error()); + if (auto r = queue.schedule ? st->bind(5, std::string_view(schedule_json)) : st->bind_null(5); !r) + return std::unexpected(r.error()); + if (auto b = st->bind(6, std::string_view(on_complete)); !b) return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + + auto stored = get(queue.queueId); + if (!stored) return std::unexpected(stored.error()); + if (!stored->has_value()) + return std::unexpected(DbError{0, "queue.upsert: row vanished after insert"}); + return **stored; +} + } // namespace velox::daemon::store diff --git a/daemon/src/store/queues.hpp b/daemon/src/store/queues.hpp index dd1c8ef..a284aea 100644 --- a/daemon/src/store/queues.hpp +++ b/daemon/src/store/queues.hpp @@ -27,6 +27,12 @@ public: // id doesn't exist. DbResult set_state(std::string_view queue_id, std::string_view state); + // "Omit queueId to create" (queue.upsert's own words) — an empty id generates one. + // taskIds is ignored (membership changes only through download.update / queue.reorder, + // per the schema's own note); a create defaults to 'stopped', a replace keeps the + // queue's current run state (queue.upsert edits config, not run state). + DbResult upsert(velox::proto::Queue queue); + private: Db& db_; }; diff --git a/daemon/tests/store_categories_queues_test.cpp b/daemon/tests/store_categories_queues_test.cpp index bd4f696..dbd376e 100644 --- a/daemon/tests/store_categories_queues_test.cpp +++ b/daemon/tests/store_categories_queues_test.cpp @@ -1,6 +1,8 @@ // store/categories + store/queues: the two D3 handlers GUI's category panel and queue // view need against a real daemon. +#include +#include #include #include "check.hpp" @@ -73,6 +75,105 @@ void run() { CHECK_EQ(ids[2], std::string("t0")); } } + + // --- Categories::upsert: create generates an id, builtin is never settable ------ + { + Categories categories(*db); + + velox::proto::Category in; + in.name = "ISOs"; + in.saveDir = "/tmp/isos"; + in.extensions = {"iso"}; + in.builtin = true; // ignored on create: a client cannot mint a builtin category + auto created = categories.upsert(in); + CHECK(created.has_value()); + if (created) { + CHECK(!created->categoryId.empty()); + CHECK(!created->builtin); + + // A replace keeps builtin=false too, and can rename/re-point. + velox::proto::Category patch = *created; + patch.name = "ISO Images"; + patch.builtin = true; // still ignored + auto replaced = categories.upsert(patch); + CHECK(replaced.has_value()); + if (replaced) { + CHECK_EQ(replaced->categoryId, created->categoryId); + CHECK_EQ(replaced->name, std::string("ISO Images")); + CHECK(!replaced->builtin); + } + + // A builtin category is untouched by remove(), and its tasks are not + // reassigned away from it — the store enforces this even without the + // dispatcher's own -32602 pre-check. + auto builtin_attempt = categories.remove("general", std::nullopt); + CHECK(builtin_attempt.has_value()); + if (builtin_attempt) CHECK(!builtin_attempt->removed); + + // remove() reassigns member tasks (default target: "general") and deletes + // the row. + Tasks tasks(*db); + TaskRow r; + r.task_id = "cat-owner"; + r.url = "https://example.com/x"; + r.save_dir = "/tmp"; + r.filename = "x"; + r.created_at = "2026-09-11T00:00:00Z"; + r.category_id = created->categoryId; + CHECK(tasks.insert(r).has_value()); + + auto removed = categories.remove(created->categoryId, std::nullopt); + CHECK(removed.has_value()); + if (removed) { + CHECK(removed->removed); + CHECK_EQ(removed->reassigned_task_ids.size(), std::size_t{1}); + CHECK_EQ(removed->reassigned_task_ids[0], std::string("cat-owner")); + } + auto owner = tasks.get("cat-owner"); + CHECK(owner.has_value() && owner->has_value()); + if (owner && *owner) + CHECK_EQ((*owner)->category_id.value_or(""), std::string("general")); + + // Gone: a second remove() finds nothing. + auto gone = categories.remove(created->categoryId, std::nullopt); + CHECK(gone.has_value()); + if (gone) CHECK(!gone->removed); + + tasks.remove("cat-owner"); + } + } + + // --- Queues::upsert: create generates an id; replace keeps the run state -------- + { + Queues queues(*db); + + velox::proto::Queue in; + in.name = "Nightly"; + in.state = velox::proto::QueueState::Running; // ignored on create: always 'stopped' + in.maxConcurrent = 3; + auto created = queues.upsert(in); + CHECK(created.has_value()); + if (created) { + CHECK(!created->queueId.empty()); + CHECK(created->state == velox::proto::QueueState::Stopped); + + CHECK(queues.set_state(created->queueId, "running").has_value()); + + velox::proto::Queue patch = *created; + patch.name = "Nightly Batch"; + patch.maxConcurrent = 5; + patch.state = velox::proto::QueueState::Stopped; // ignored on replace too + auto replaced = queues.upsert(patch); + CHECK(replaced.has_value()); + if (replaced) { + CHECK_EQ(replaced->name, std::string("Nightly Batch")); + CHECK_EQ(replaced->maxConcurrent, std::int64_t{5}); + // Run state survived the config edit — still 'running' from set_state above, + // not reset by the payload's (ignored) 'stopped'. + CHECK(replaced->state == velox::proto::QueueState::Running); + } + } + } } TEST_MAIN() From cf9e226e61cf8f3cd54a8244cbbb6f87cb5786c4 Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:46:43 +0400 Subject: [PATCH 5/5] =?UTF-8?q?daemon:=20D1=20=E2=80=94=20checked,=20not?= =?UTF-8?q?=20attempted;=20document=20why?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit libdbus-1-dev / libsystemd-dev have no headers installed in this build environment (only the runtime .so's — apt-cache policy confirms libdbus-1-dev is available but not installed). A real org.freedesktop.Notifications-backed PairingApprover needs one of those linked into veloxd, which is a new build dependency for daemon/CMakeLists.txt and, since packaging manifests would need to know about it too, a decision to surface rather than reach for silently mid-session. PairingApprover::approve() is also still synchronous by shape — its own doc comment already says the real approver "will run async and is not this shape." The async pattern this session built for download.probe (rpc::TaskActionPort + the server-layer deferred-reply special-case in uds_server.cpp/ws_server.cpp) is the right shape to reuse once there's a real implementation to justify reshaping the interface; doing that with nothing behind it yet would just be churn. Left EnvAutoApprover in place rather than hand-roll a D-Bus wire client to route around the missing headers — a broken pairing approver is a worse outcome than an honest, already-documented stub. Findings recorded in deferrals.md for whoever picks this up once the dependency is available and approved. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 1 + 1 file changed, 1 insertion(+) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 588553a..c509c95 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -6,6 +6,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | 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) | +| — | **D1, checked this pass, not attempted:** `libdbus-1-dev` (or `libsystemd-dev` for `sd-bus`) has no headers installed in this build environment — only the runtime `.so`s (`dpkg -l`/`apt-cache policy` confirm `libdbus-1-3` present, `libdbus-1-dev` not, "Candidate" available but not installed). A real notification-backed approver needs one of those linked into `veloxd`, which is a new build dependency for `daemon/CMakeLists.txt` (`find_package`/`pkg_check_modules`) and — since packaging manifests need to know about it too — arguably a decision to surface rather than something to reach for silently mid-session. `PairingApprover::approve()` is also still synchronous by shape (its own doc comment already says so: "the real notification-backed approver will run async and is not this shape") — swapping it for the async pattern this session built for `download.probe` (`rpc::TaskActionPort` + the server-layer deferred-reply special-case) is the right shape once there's a real implementation to justify the churn; reshaping the interface with nothing behind it yet would just be churn. Left `EnvAutoApprover` in place rather than build a fragile hand-rolled D-Bus wire client to avoid the missing headers — a broken pairing approver is worse than an honest stub. | `rpc/pairing.hpp` | missing dev headers + an undiscussed new dependency | once `libdbus-1-dev`/`libsystemd-dev` is available and the dependency is approved | | ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | | D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet, or (`settings.*`) sound but large — see the note below. `category.list/upsert/remove`, `queue.list/upsert/start/stop`, `download.remove/addBatch/provideAuth` are done | Per method, as each wires to the store/scheduler | | — | **`settings.get`/`settings.set` specifically, not started:** `proto::Settings` is a flat struct of ~43 `std::optional` fields, one per `SettingKey` (~50 keys) in `Settings.schema.json`; `store::Settings` already has `get_raw`/`set_raw`/`overrides` keyed by the same dotted strings the JSON uses (`"connection.maxSegmentsPerDownload"`, …). The handlers are a mechanical field <-> key <-> JSON-type mapping table in both directions (get: row-or-default -> struct field; set: struct field -> validate against the key's schema type -> `set_raw`, collecting `changed`) — real work, just long and repetitive rather than hard. Left alone this pass rather than rushed; every other read of settings in this codebase already goes through `store::Settings`'s typed helpers directly (`reload_config`, `on_download_add`'s segment default, `capture.offer`'s allowed roots when that lands), so nothing downstream is blocked on the RPC surface existing. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}` | the mapping table is genuinely large, not genuinely hard | its own pass |