sami/vdm
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

merge: lane/daemon

Browse Source
This commit is contained in:
sami
2026-09-12 11:05:07 +04:00
parent ff721f4065 cf9e226e61
commit b578e6de1a
25 changed files with 1483 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/CMakeLists.txt
+10 -1
View File
@@ -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
@@ -87,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 -------------------------------------------------------
daemon/docs/deferrals.md
+9 -4
View File
@@ -6,10 +6,15 @@ 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`) |
| 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 |
| — | **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<T>` 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 |
| ~~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 | 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. |
daemon/src/main.cpp
+11 -33
View File
@@ -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<socklen_t>(offsetof(sockaddr_un, sun_path) + 1 + name.size());
if (::bind(fd, reinterpret_cast<sockaddr*>(&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;
@@ -125,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(); });
});
@@ -183,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";
@@ -197,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";
daemon/src/rpc/dispatcher.cpp
+256 -30
View File
@@ -1,5 +1,6 @@
#include "rpc/dispatcher.hpp"
#include <filesystem>
#include <random>
#include <string>
@@ -28,6 +29,29 @@ proto::HandlerResult<T> 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<std::string>& task_ids,
const std::function<velox::daemon::rpc::TaskActionPort::Result(const std::string&)>&
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;
@@ -118,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<proto::DownloadAddResult>
VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
proto::HandlerResult<proto::DownloadAddResult> VeloxDispatcher::add_one(
const proto::DownloadSpec& spec) {
store::Settings settings(db_);
std::string save_dir = spec.saveDir && !spec.saveDir->empty()
@@ -152,10 +180,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;
@@ -194,6 +231,11 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
return r;
}
proto::HandlerResult<proto::DownloadAddResult>
VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
return add_one(spec);
}
// --- everything else : not implemented until the store and scheduler land -------------
proto::HandlerResult<proto::CaptureRules>
@@ -216,20 +258,103 @@ VeloxDispatcher::on_category_list(const proto::CategoryListParams&) {
return r;
}
proto::HandlerResult<proto::CategoryRemoveResult>
VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) {
return not_implemented<proto::CategoryRemoveResult>("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<proto::CategoryUpsertResult>
VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) {
return not_implemented<proto::CategoryUpsertResult>("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<std::string> 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<proto::DownloadAddBatchResult>
VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) {
return not_implemented<proto::DownloadAddBatchResult>("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<std::int64_t>(i), added.error().code,
added.error().message});
}
}
return r;
}
proto::HandlerResult<proto::BulkTaskResult>
VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) {
return not_implemented<proto::BulkTaskResult>("download.cancel");
VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) {
if (!actions_) return not_implemented<proto::BulkTaskResult>("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<proto::TaskDetail>
VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) {
@@ -264,32 +389,84 @@ VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) {
return d;
}
proto::HandlerResult<proto::BulkTaskResult>
VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) {
return not_implemented<proto::BulkTaskResult>("download.pause");
VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams& params) {
if (!actions_) return not_implemented<proto::BulkTaskResult>("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<proto::DownloadProbeResult>
VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) {
return not_implemented<proto::DownloadProbeResult>("download.probe");
}
proto::HandlerResult<proto::DownloadProvideAuthResult>
VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams&) {
return not_implemented<proto::DownloadProvideAuthResult>("download.provideAuth");
VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams& params) {
if (!actions_) return not_implemented<proto::DownloadProvideAuthResult>("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<proto::DownloadRefreshUrlResult>
VeloxDispatcher::on_download_refreshUrl(const proto::DownloadRefreshUrlParams&) {
return not_implemented<proto::DownloadRefreshUrlResult>("download.refreshUrl");
}
proto::HandlerResult<proto::DownloadRemoveResult>
VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) {
return not_implemented<proto::DownloadRemoveResult>("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<proto::BulkTaskResult>
VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams&) {
return not_implemented<proto::BulkTaskResult>("download.resume");
VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) {
if (!actions_) return not_implemented<proto::BulkTaskResult>("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<proto::BulkTaskResult>
VeloxDispatcher::on_download_start(const proto::DownloadStartParams&) {
return not_implemented<proto::BulkTaskResult>("download.start");
VeloxDispatcher::on_download_start(const proto::DownloadStartParams& params) {
if (!actions_) return not_implemented<proto::BulkTaskResult>("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<proto::TaskSummary>
VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams&) {
@@ -337,16 +514,65 @@ VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) {
return not_implemented<proto::QueueReorderResult>("queue.reorder");
}
proto::HandlerResult<proto::QueueStartResult>
VeloxDispatcher::on_queue_start(const proto::QueueStartParams&) {
return not_implemented<proto::QueueStartResult>("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<proto::QueueStopResult>
VeloxDispatcher::on_queue_stop(const proto::QueueStopParams&) {
return not_implemented<proto::QueueStopResult>("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<proto::QueueUpsertResult>
VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) {
return not_implemented<proto::QueueUpsertResult>("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<proto::RulesListResult>
VeloxDispatcher::on_rules_list(const proto::RulesListParams&) {
daemon/src/rpc/dispatcher.hpp
+17 -1
View File
@@ -15,6 +15,7 @@
#include <functional>
#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.
@@ -107,8 +116,15 @@ 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<velox::proto::DownloadAddResult> add_one(
const velox::proto::DownloadSpec& spec);
velox::daemon::store::Db& db_;
EventHub& hub_;
TaskActionPort* actions_;
std::function<void()> on_mutation_;
};
daemon/src/rpc/single_instance.cpp
+38
View File
@@ -0,0 +1,38 @@
#include "rpc/single_instance.hpp"
#include <cstddef>
#include <cstring>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#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<socklen_t>(offsetof(sockaddr_un, sun_path) + 1 + name.size());
if (::bind(fd, reinterpret_cast<sockaddr*>(&addr), len) != 0) {
::close(fd);
return -1;
}
return fd;
}
} // namespace velox::daemon::rpc
daemon/src/rpc/single_instance.hpp
+27
View File
@@ -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/<uid>/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-<euid>" 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 <string>
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
daemon/src/rpc/task_action_port.hpp
+71
View File
@@ -0,0 +1,71 @@
#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 <functional>
#include <string>
#include <vector>
#include "velox_proto.hpp"
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<std::string> 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:
// 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<void(velox::proto::HandlerResult<velox::proto::DownloadProbeResult>)>
done) = 0;
};
} // namespace velox::daemon::rpc
daemon/src/rpc/uds_server.cpp
+38 -2
View File
@@ -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<proto::DownloadProbeParams>(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<proto::DownloadProbeResult> 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);
daemon/src/rpc/uds_server.hpp
+13 -1
View File
@@ -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
daemon/src/rpc/ws_server.cpp
+38 -1
View File
@@ -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<proto::DownloadProbeParams>(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<proto::DownloadProbeResult> 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);
daemon/src/rpc/ws_server.hpp
+11 -1
View File
@@ -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_;
daemon/src/sched/scheduler.cpp
+187 -1
View File
@@ -6,6 +6,7 @@
#include <nlohmann/json.hpp>
#include "sched/schedule_window.hpp"
#include "store/categories.hpp"
#include "store/segments.hpp"
#include "store/settings.hpp"
#include "store/tasks.hpp"
@@ -291,7 +292,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<std::string> 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 +482,180 @@ std::vector<Scheduler::ProgressRow> 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<std::string> 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<std::string> 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;
}
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),
// 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<char>(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<void(proto::HandlerResult<proto::DownloadProbeResult>)> 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<vdm::net::ProbeResult> 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<std::int64_t>(*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
daemon/src/sched/scheduler.hpp
+50 -1
View File
@@ -25,6 +25,7 @@
#include <vector>
#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<std::int64_t> 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,49 @@ public:
};
std::vector<ProgressRow> 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<std::string> 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
// 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<void(velox::proto::HandlerResult<velox::proto::DownloadProbeResult>)>
done) override;
// Diagnostics / tests.
std::optional<std::string> wire_id_of(vdm::TaskId id) const;
std::optional<vdm::TaskId> engine_id_of(const std::string& wire_id) const;
daemon/src/store/categories.cpp
+127 -12
View File
@@ -1,11 +1,35 @@
#include "store/categories.hpp"
#include <sqlite3.h>
#include <cstdio>
#include <random>
#include <string>
#include <nlohmann/json.hpp>
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<std::string>());
}
c.builtin = st.column_int(4) != 0;
return c;
}
} // namespace
DbResult<std::vector<proto::Category>> Categories::list() {
auto st = db_.prepare(
"SELECT category_id, name, save_dir, extensions, builtin FROM categories "
@@ -17,20 +41,111 @@ DbResult<std::vector<proto::Category>> 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<std::string>());
}
c.builtin = st->column_int(4) != 0;
out.push_back(std::move(c));
out.push_back(project_row(*st));
}
return out;
}
DbResult<std::optional<proto::Category>> 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<proto::Category>{};
return std::optional<proto::Category>{project_row(*st)};
}
DbResult<proto::Category> 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<std::uint64_t> d;
char buf[17];
std::snprintf(buf, sizeof(buf), "%016llx", static_cast<unsigned long long>(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<std::int64_t>(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::RemoveResult> Categories::remove(
std::string_view category_id, const std::optional<std::string>& reassign_to) {
RemoveResult out;
const std::string target = reassign_to.value_or("general");
// Db::transaction only threads a DbResult<void> lambda; `out` is filled in-place and
// returned once the transaction (which may still fail and roll back) succeeds.
auto txn = db_.transaction([&]() -> DbResult<void> {
{
// 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
daemon/src/store/categories.hpp
+24
View File
@@ -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 <optional>
#include <string>
#include <vector>
#include "store/sqlite.hpp"
@@ -16,6 +22,24 @@ public:
explicit Categories(Db& db) : db_(db) {}
DbResult<std::vector<velox::proto::Category>> list();
DbResult<std::optional<velox::proto::Category>> 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<velox::proto::Category> 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<std::string> reassigned_task_ids;
};
DbResult<RemoveResult> remove(std::string_view category_id,
const std::optional<std::string>& reassign_to);
private:
Db& db_;
daemon/src/store/migrations/0003_start_mode_contract_values.sql
+91
View File
@@ -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);
daemon/src/store/queues.cpp
+110 -26
View File
@@ -1,14 +1,53 @@
#include "store/queues.hpp"
#include <sqlite3.h>
#include <cstdio>
#include <random>
#include <string>
#include <nlohmann/json.hpp>
namespace velox::daemon::store {
namespace proto = velox::proto;
namespace {
// 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<proto::Queue> 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<proto::Schedule>(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());
if (auto b = ts->bind(1, std::string_view(q.queueId)); !b) return std::unexpected(b.error());
std::vector<std::string> 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<std::vector<proto::Queue>> 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<proto::Queue> out;
@@ -16,33 +55,78 @@ DbResult<std::vector<proto::Queue>> 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<proto::Schedule>(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<std::string> 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<std::optional<proto::Queue>> Queues::get(std::string_view queue_id) {
auto st = db_.prepare(
"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();
if (!row) return std::unexpected(row.error());
if (!*row) return std::optional<proto::Queue>{};
auto q = project_row(db_, *st);
if (!q) return std::unexpected(q.error());
return std::optional<proto::Queue>{std::move(*q)};
}
DbResult<bool> 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;
}
DbResult<proto::Queue> Queues::upsert(proto::Queue queue) {
if (queue.queueId.empty()) {
std::random_device rd;
std::uniform_int_distribution<std::uint64_t> d;
char buf[17];
std::snprintf(buf, sizeof(buf), "%016llx", static_cast<unsigned long long>(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
daemon/src/store/queues.hpp
+15
View File
@@ -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 <optional>
#include <string_view>
#include <vector>
#include "store/sqlite.hpp"
@@ -18,6 +20,19 @@ public:
DbResult<std::vector<velox::proto::Queue>> list();
// nullopt (not an error) if no queue has this id.
DbResult<std::optional<velox::proto::Queue>> 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<bool> 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<velox::proto::Queue> upsert(velox::proto::Queue queue);
private:
Db& db_;
};
daemon/src/store/tasks.hpp
+1 -1
View File
@@ -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<std::string> effective_url;
daemon/tests/CMakeLists.txt
+1
View File
@@ -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)
daemon/tests/sched_scheduler_test.cpp
+147
View File
@@ -259,6 +259,153 @@ 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());
}
// --- 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<vdm::net::ProbeResult>{pr};
velox::proto::DownloadProbeParams params;
params.url = "https://cdn.example/movie.mp4";
std::optional<velox::proto::HandlerResult<velox::proto::DownloadProbeResult>> 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<vdm::net::ProbeResult>{err};
std::optional<velox::proto::HandlerResult<velox::proto::DownloadProbeResult>> 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()
daemon/tests/single_instance_test.cpp
+40
View File
@@ -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 <unistd.h>
#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()
daemon/tests/store_categories_queues_test.cpp
+101
View File
@@ -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 <algorithm>
#include <optional>
#include <string>
#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()
daemon/tests/store_migrations_test.cpp
+50
View File
@@ -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:");