From e0edf7a0842a1c5f60f5360a28511dcbfe874608 Mon Sep 17 00:00:00 2001 From: sami Date: Sat, 12 Sep 2026 13:42:45 +0400 Subject: [PATCH 1/3] daemon: file capture.offer/getRules in deferrals (D7/D8); close settings.get/set (D9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebased lane/daemon onto main first — fast-forward, no conflicts (lane/daemon was already fully merged; main has since taken lane/gui, lane/pkg-qa, lane/proto). 1. capture.offer and capture.getRules were stubs, buried inside D3's generic stub list with no flag for how load-bearing capture.offer is: it's a CLAUDE.md §4 non-negotiable (750ms deadline, fails open) and AGENT-DAEMON.md build step 6, and against the real daemon it 500s to -32603 every time — the extension captures nothing, not "falls through to Firefox on a slow path." Split into their own D7/D8 rows so they stop being invisible. 2. settings.get/settings.set (D9, closed). Four pointer-to-member tables in dispatcher.cpp (bool / ranged int / plain string / string array), five enum-typed keys handled individually since parse_XXX already validates those — covers all 43 SettingKeys without ~40 repetitive hand blocks. store::Settings::kDefaults grew from 14 entries to 43 (the schema itself carries no "default" keyword anywhere, so these are hand-chosen — conservative for the ones ADR 0012 doesn't speak to; capture.monitoredExtensions defaults to the union of every builtin category's extensions rather than an arbitrary list of its own). settings.set validates every field before writing any of them: numeric min/max (hand-checked — the generated parser only checks JSON type, not schema constraints) and saveTo.* paths via fs::resolve_target/canonicalize_root (-32011) — allowedRoots entries checked as roots in their own right, defaultDir/tempDir checked as paths resolving inside the (possibly just-updated, same call) root list. Reports exactly the keys whose effective value actually changed, publishes event.settings.changed with that list, and calls the scheduler's reload_config() when a connection.* key took effect — live, not on next restart. Verified against real veloxd: all 43 keys round-trip with defaults, a keys subset filters, an out-of-range value rejects the whole call, a bad saveTo.defaultDir is -32011, allowedRoots+defaultDir set together cross-validate against the new roots, event.settings.changed fires over a live subscription. New dispatcher_settings_test covers the same ground without a socket. Full ctest: 53/53 (excluding a pre-existing, unrelated conformance failure — see below). Also found, not fixed (not this lane): tests/conformance/run.sh's live-veloxd leg fails "pairing failed: no token issued" reproducibly, isolated, on main before any of this session's changes — it launches the isolated veloxd without VELOX_PAIR_AUTO=1, so EnvAutoApprover denies every session.pair and the WS leg's session.hello never gets a token. tests/conformance/ is PROTO/QA-owned; flagging rather than editing across the lane boundary. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 6 +- daemon/src/rpc/dispatcher.cpp | 291 +++++++++++++++++++++- daemon/src/rpc/task_action_port.hpp | 8 + daemon/src/sched/scheduler.hpp | 6 + daemon/src/store/settings.cpp | 69 ++++- daemon/src/store/settings.hpp | 1 + daemon/tests/CMakeLists.txt | 1 + daemon/tests/dispatcher_settings_test.cpp | 142 +++++++++++ 8 files changed, 509 insertions(+), 15 deletions(-) create mode 100644 daemon/tests/dispatcher_settings_test.cpp diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index c509c95..014fd51 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -5,11 +5,13 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | What | Where | Why deferred | Closes when | |---|---|---|---|---| +| **D7** | **`capture.offer` → `-32603`.** This is a CLAUDE.md §4 non-negotiable ("Capture fails open... the extension gives up" — but a fail-open answer still has to arrive; a hard error is not a substitute for `ignore`) and AGENT-DAEMON.md build step 6 ("`capture.offer` must answer within 750 ms, always"). Against the real daemon right now, every offer 500s, so the extension captures nothing at all — not "falls through to Firefox on a slow path," genuinely nothing, because the extension is waiting on a promise that never resolves with a decision it can act on. This was buried inside D3's stub list ("...`capture.*`") with no flag for how load-bearing it is; called out on its own row now. | `rpc/dispatcher.cpp` | rules table has no matcher yet, no dedup-against-active-tasks, no category-folder resolution | build step 6 | +| **D8** | `capture.getRules` → `-32603`. The extension mirrors this into its own capture policy on connect (docs/05 §4) so the two can never disagree about what should be intercepted; until this returns something real the extension runs with an empty/stale policy. Also buried in D3's list before this pass. | `rpc/dispatcher.cpp` | same rules-engine gap as D7 | alongside D7 | | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | | — | **D1, checked this pass, not attempted:** `libdbus-1-dev` (or `libsystemd-dev` for `sd-bus`) has no headers installed in this build environment — only the runtime `.so`s (`dpkg -l`/`apt-cache policy` confirm `libdbus-1-3` present, `libdbus-1-dev` not, "Candidate" available but not installed). A real notification-backed approver needs one of those linked into `veloxd`, which is a new build dependency for `daemon/CMakeLists.txt` (`find_package`/`pkg_check_modules`) and — since packaging manifests need to know about it too — arguably a decision to surface rather than something to reach for silently mid-session. `PairingApprover::approve()` is also still synchronous by shape (its own doc comment already says so: "the real notification-backed approver will run async and is not this shape") — swapping it for the async pattern this session built for `download.probe` (`rpc::TaskActionPort` + the server-layer deferred-reply special-case) is the right shape once there's a real implementation to justify the churn; reshaping the interface with nothing behind it yet would just be churn. Left `EnvAutoApprover` in place rather than build a fragile hand-rolled D-Bus wire client to avoid the missing headers — a broken pairing approver is worse than an honest stub. | `rpc/pairing.hpp` | missing dev headers + an undiscussed new dependency | once `libdbus-1-dev`/`libsystemd-dev` is available and the dependency is approved | | ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | -| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet, or (`settings.*`) sound but large — see the note below. `category.list/upsert/remove`, `queue.list/upsert/start/stop`, `download.remove/addBatch/provideAuth` are done | Per method, as each wires to the store/scheduler | -| — | **`settings.get`/`settings.set` specifically, not started:** `proto::Settings` is a flat struct of ~43 `std::optional` fields, one per `SettingKey` (~50 keys) in `Settings.schema.json`; `store::Settings` already has `get_raw`/`set_raw`/`overrides` keyed by the same dotted strings the JSON uses (`"connection.maxSegmentsPerDownload"`, …). The handlers are a mechanical field <-> key <-> JSON-type mapping table in both directions (get: row-or-default -> struct field; set: struct field -> validate against the key's schema type -> `set_raw`, collecting `changed`) — real work, just long and repetitive rather than hard. Left alone this pass rather than rushed; every other read of settings in this codebase already goes through `store::Settings`'s typed helpers directly (`reload_config`, `on_download_add`'s segment default, `capture.offer`'s allowed roots when that lands), so nothing downstream is blocked on the RPC surface existing. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}` | the mapping table is genuinely large, not genuinely hard | its own pass | +| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet | Per method, as each wires to the store/scheduler | +| ~~D9~~ | **Closed — `settings.get`/`settings.set`.** The field <-> `SettingKey` <-> JSON-type mapping is four pointer-to-member tables in `dispatcher.cpp` (one per C++ field type: bool, ranged int, plain string, string array) plus five enum-typed keys handled individually (`parse_XXX` already validates those); every one of the 43 `SettingKey`s now has a real default (`store::Settings::kDefaults` grew from 14 entries to 43 — `capture.monitoredExtensions`'s default is the union of every builtin category's extensions, so the two never drift apart). `settings.get` honors `keys: null` = everything. `settings.set` validates every field *before* writing any of them (numeric min/max — the schema itself carries none of this, so it's hand-checked against each key's documented range; `-32602` names the offending key, its value, and its bounds) and validates `saveTo.*` paths against `fs::resolve_target`/`canonicalize_root` (`-32011`) — `saveTo.allowedRoots` entries are checked as roots in their own right, `saveTo.defaultDir`/`.tempDir` are checked as paths resolving *inside* the (possibly, in the same call, just-updated) root list. Reports exactly the keys whose *effective* value actually changed (a `set` to the value already in effect reports `changed: []`, not the key), publishes `event.settings.changed` with that same list, and calls `TaskActionPort::apply_settings_reload()` (-> `Scheduler::reload_config()`) when any `connection.*` key took effect, live rather than waiting for a restart. Verified against real `veloxd`: all 43 keys round-trip with sane defaults, a `keys` subset filters correctly, an out-of-range value is rejected with nothing else in the same call landing, `saveTo.defaultDir` outside every allowed root is `-32011`, setting `allowedRoots` and `defaultDir` together cross-validates against the *new* roots, and `event.settings.changed` fires over a live subscription. New `dispatcher_settings_test` covers the same ground without a socket. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}`, `rpc/task_action_port.hpp`, `sched/scheduler.hpp` | — | done | | ~~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 | diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index c88d8be..7490e33 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -1,6 +1,9 @@ #include "rpc/dispatcher.hpp" +#include +#include #include +#include #include #include @@ -102,6 +105,115 @@ std::string filename_from_url(std::string_view url) { return out; } +// --- settings.get / settings.set : the field <-> SettingKey <-> JSON-type mapping ------ +// +// One table per C++ field type in proto::Settings, keyed by pointer-to-member so get/set +// each need one small loop per type instead of ~43 repetitive hand-written blocks. Every +// SettingKey appears in exactly one table (or, for the five enum-typed keys, is handled +// individually just below — parse_XXX already does their validation, so there is nothing +// generic left for those five to share). store::Settings::get_raw/get_bool/get_int/ +// get_string/get_string_array already do the row-or-default fallback; these tables just +// say which key goes with which field and, for the numeric ones, its valid range (schema +// minimum/maximum — NOT enforced by the generated parser, which only checks JSON type). + +struct BoolSetting { + proto::SettingKey key; + std::optional proto::Settings::* field; +}; +constexpr BoolSetting kBoolSettings[] = { + {proto::SettingKey::GeneralLaunchOnLogin, &proto::Settings::general_launchOnLogin}, + {proto::SettingKey::GeneralMinimizeToTray, &proto::Settings::general_minimizeToTray}, + {proto::SettingKey::GeneralShowDropTarget, &proto::Settings::general_showDropTarget}, + {proto::SettingKey::GeneralConfirmOnExit, &proto::Settings::general_confirmOnExit}, + {proto::SettingKey::GeneralCheckForUpdates, &proto::Settings::general_checkForUpdates}, + {proto::SettingKey::CaptureEnabled, &proto::Settings::capture_enabled}, + {proto::SettingKey::SaveToCreateSubfolderPerSite, &proto::Settings::saveTo_createSubfolderPerSite}, + {proto::SettingKey::DownloadsSpeedLimitEnabled, &proto::Settings::downloads_speedLimitEnabled}, + {proto::SettingKey::DownloadsVerifyChecksums, &proto::Settings::downloads_verifyChecksums}, + {proto::SettingKey::SoundsEnabled, &proto::Settings::sounds_enabled}, +}; + +struct IntSetting { + proto::SettingKey key; + std::optional proto::Settings::* field; + std::int64_t min; + std::int64_t max; +}; +constexpr std::int64_t kNoMax = std::numeric_limits::max(); +constexpr IntSetting kIntSettings[] = { + {proto::SettingKey::CaptureMinSizeBytes, &proto::Settings::capture_minSizeBytes, 0, kNoMax}, + {proto::SettingKey::ConnectionMaxSegmentsPerDownload, + &proto::Settings::connection_maxSegmentsPerDownload, 1, 32}, + {proto::SettingKey::ConnectionBufferBytes, &proto::Settings::connection_bufferBytes, 65536, + 16777216}, + {proto::SettingKey::ConnectionMaxTotalBufferBytes, + &proto::Settings::connection_maxTotalBufferBytes, 16777216, 2147483648}, + {proto::SettingKey::ConnectionMaxActiveSegments, &proto::Settings::connection_maxActiveSegments, + 1, 256}, + {proto::SettingKey::ConnectionMaxConcurrentDownloads, + &proto::Settings::connection_maxConcurrentDownloads, 1, 64}, + {proto::SettingKey::ConnectionTimeoutSec, &proto::Settings::connection_timeoutSec, 1, 3600}, + {proto::SettingKey::ConnectionMaxRetries, &proto::Settings::connection_maxRetries, 0, 100}, + {proto::SettingKey::ConnectionRetryBackoffSec, &proto::Settings::connection_retryBackoffSec, 0, + 3600}, + {proto::SettingKey::DownloadsSpeedLimitBps, &proto::Settings::downloads_speedLimitBps, 0, kNoMax}, + {proto::SettingKey::ProxyPort, &proto::Settings::proxy_port, 1, 65535}, +}; + +struct StrSetting { + proto::SettingKey key; + std::optional proto::Settings::* field; +}; +// saveTo.defaultDir / saveTo.tempDir are here too (read/write is identical to any other +// string); on_settings_set gives them one extra pass validating the path itself before +// either is written. +constexpr StrSetting kStrSettings[] = { + {proto::SettingKey::GeneralLanguage, &proto::Settings::general_language}, + {proto::SettingKey::SaveToDefaultDir, &proto::Settings::saveTo_defaultDir}, + {proto::SettingKey::SaveToTempDir, &proto::Settings::saveTo_tempDir}, + {proto::SettingKey::DownloadsVirusScanCommand, &proto::Settings::downloads_virusScanCommand}, + {proto::SettingKey::DownloadsPostDownloadCommand, &proto::Settings::downloads_postDownloadCommand}, + {proto::SettingKey::ProxyHost, &proto::Settings::proxy_host}, + {proto::SettingKey::ProxyUsername, &proto::Settings::proxy_username}, + {proto::SettingKey::ProxyPacUrl, &proto::Settings::proxy_pacUrl}, + {proto::SettingKey::SoundsOnComplete, &proto::Settings::sounds_onComplete}, + {proto::SettingKey::SoundsOnQueueComplete, &proto::Settings::sounds_onQueueComplete}, + {proto::SettingKey::SoundsOnError, &proto::Settings::sounds_onError}, +}; + +struct StrArrSetting { + proto::SettingKey key; + std::optional> proto::Settings::* field; +}; +// saveTo.allowedRoots is here too; same deal as the two directory strings above, except +// every entry is checked (it IS the root list, not a path resolved against it). +constexpr StrArrSetting kStrArrSettings[] = { + {proto::SettingKey::CaptureMonitoredExtensions, &proto::Settings::capture_monitoredExtensions}, + {proto::SettingKey::CaptureMonitoredMimeTypes, &proto::Settings::capture_monitoredMimeTypes}, + {proto::SettingKey::CaptureExcludedHosts, &proto::Settings::capture_excludedHosts}, + {proto::SettingKey::CaptureAutoStartTypes, &proto::Settings::capture_autoStartTypes}, + {proto::SettingKey::ProxyBypassHosts, &proto::Settings::proxy_bypassHosts}, + {proto::SettingKey::SaveToAllowedRoots, &proto::Settings::saveTo_allowedRoots}, +}; + +// Whether `key` is in `wanted` (settings.get's own params.keys, nullopt meaning "all" per +// the schema's "keys null means everything"). +bool settings_key_wanted(const std::optional>& wanted, + proto::SettingKey key) { + if (!wanted) return true; + return std::find(wanted->begin(), wanted->end(), key) != wanted->end(); +} + +// Whether writing `key` would actually change the effective value (row-or-default) — +// settings.set / event.settings.changed report exactly the keys that "took effect", not +// every key the caller merely mentioned. +bool settings_value_changed(store::Settings& settings, proto::SettingKey key, + const std::string& new_json) { + auto old = settings.get_raw(proto::to_string(key)); + const std::string old_json = (old && *old) ? **old : std::string(); + return old_json != new_json; +} + } // namespace // --- session.* : handled in the server layer, unreachable here in the running daemon --- @@ -591,12 +703,183 @@ VeloxDispatcher::on_schedule_set(const proto::ScheduleSetParams&) { return not_implemented("schedule.set"); } proto::HandlerResult -VeloxDispatcher::on_settings_get(const proto::SettingsGetParams&) { - return not_implemented("settings.get"); +VeloxDispatcher::on_settings_get(const proto::SettingsGetParams& params) { + store::Settings settings(db_); + proto::Settings out; + + for (const auto& s : kBoolSettings) + if (settings_key_wanted(params.keys, s.key)) out.*(s.field) = settings.get_bool(proto::to_string(s.key)); + for (const auto& s : kIntSettings) + if (settings_key_wanted(params.keys, s.key)) out.*(s.field) = settings.get_int(proto::to_string(s.key)); + for (const auto& s : kStrSettings) + if (settings_key_wanted(params.keys, s.key)) out.*(s.field) = settings.get_string(proto::to_string(s.key)); + for (const auto& s : kStrArrSettings) + if (settings_key_wanted(params.keys, s.key)) + out.*(s.field) = settings.get_string_array(proto::to_string(s.key)); + + // The five enum-typed keys: get_string() already returns the raw-or-default JSON + // string content (quotes stripped); parse_XXX validates it against the same set the + // schema's enum names. A stored value that somehow doesn't parse (should not happen — + // set_raw always goes through the same parser) leaves the field unset rather than + // guessing, which is a legal Settings shape (every property is optional). + if (settings_key_wanted(params.keys, proto::SettingKey::CaptureBypassModifier)) + if (auto v = proto::parse_BypassModifier(settings.get_string("capture.bypassModifier"))) + out.capture_bypassModifier = *v; + if (settings_key_wanted(params.keys, proto::SettingKey::SaveToFileExistsPolicy)) + if (auto v = proto::parse_SettingsSaveToFileExistsPolicy( + settings.get_string("saveTo.fileExistsPolicy"))) + out.saveTo_fileExistsPolicy = *v; + if (settings_key_wanted(params.keys, proto::SettingKey::ConnectionPreset)) + if (auto v = proto::parse_SettingsConnectionPreset(settings.get_string("connection.preset"))) + out.connection_preset = *v; + if (settings_key_wanted(params.keys, proto::SettingKey::DownloadsDuplicatePolicy)) + if (auto v = proto::parse_SettingsDownloadsDuplicatePolicy( + settings.get_string("downloads.duplicatePolicy"))) + out.downloads_duplicatePolicy = *v; + if (settings_key_wanted(params.keys, proto::SettingKey::ProxyMode)) + if (auto v = proto::parse_SettingsProxyMode(settings.get_string("proxy.mode"))) + out.proxy_mode = *v; + + proto::SettingsGetResult r; + r.values = std::move(out); + return r; } + proto::HandlerResult -VeloxDispatcher::on_settings_set(const proto::SettingsSetParams&) { - return not_implemented("settings.set"); +VeloxDispatcher::on_settings_set(const proto::SettingsSetParams& params) { + const proto::Settings& v = params.values; + store::Settings settings(db_); + + // --- pass 1: validate everything before writing anything (no partial update on a + // rejected call) ------------------------------------------------------------------ + for (const auto& s : kIntSettings) { + if (!(v.*(s.field))) continue; + const auto val = *(v.*(s.field)); + if (val < s.min || val > s.max) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, + "value out of range for " + std::string(proto::to_string(s.key)), + nlohmann::json{{"key", std::string(proto::to_string(s.key))}, {"value", val}, + {"min", s.min}, {"max", s.max}}}); + } + } + + // Directory keys: saveTo.allowedRoots entries are checked as roots in their own + // right; saveTo.defaultDir / saveTo.tempDir are checked as paths that must resolve + // *inside* the (possibly, in this same call, just-updated) root list — the same rule + // download.add's own saveDir is held to ("every write target is canonicalized and + // must resolve inside one of these", Settings.schema.json's own words for + // saveTo.allowedRoots). + const std::vector effective_roots = + v.saveTo_allowedRoots ? *v.saveTo_allowedRoots : settings.get_string_array("saveTo.allowedRoots"); + std::vector canon_roots; + for (const auto& r : effective_roots) { + auto c = fs::canonicalize_root(r); + if (!c) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidPath, "not a writable directory", + nlohmann::json{{"key", "saveTo.allowedRoots"}, {"path", r}}}); + } + canon_roots.push_back(*c); + } + + auto check_dir = [&](const std::string& dir, + const char* key_name) -> std::optional { + if (dir.empty()) return std::nullopt; // "" means "unset / use the default" + auto target = fs::resolve_target(expand_tilde(dir), ".settings-marker", canon_roots); + if (!target) { + return proto::HandlerError{proto::ErrorCode::InvalidPath, target.error().message, + nlohmann::json{{"key", key_name}, {"path", dir}}}; + } + return std::nullopt; + }; + if (v.saveTo_defaultDir) { + if (auto e = check_dir(*v.saveTo_defaultDir, "saveTo.defaultDir")) return std::unexpected(*e); + } + if (v.saveTo_tempDir) { + if (auto e = check_dir(*v.saveTo_tempDir, "saveTo.tempDir")) return std::unexpected(*e); + } + + // --- pass 2: write, tracking which keys actually took effect ---------------------- + std::vector changed; + bool touched_connection = false; + + auto apply = [&](proto::SettingKey key, const std::string& json_text) { + if (settings_value_changed(settings, key, json_text)) changed.push_back(key); + (void)settings.set_raw(proto::to_string(key), json_text); + }; + + for (const auto& s : kBoolSettings) + if (v.*(s.field)) apply(s.key, *(v.*(s.field)) ? "true" : "false"); + for (const auto& s : kIntSettings) { + if (!(v.*(s.field))) continue; + apply(s.key, std::to_string(*(v.*(s.field)))); + touched_connection = touched_connection || proto::to_string(s.key).starts_with("connection."); + } + for (const auto& s : kStrSettings) + if (v.*(s.field)) apply(s.key, nlohmann::json(*(v.*(s.field))).dump()); + for (const auto& s : kStrArrSettings) + if (v.*(s.field)) apply(s.key, nlohmann::json(*(v.*(s.field))).dump()); + + if (v.capture_bypassModifier) + apply(proto::SettingKey::CaptureBypassModifier, + nlohmann::json(std::string(proto::to_string(*v.capture_bypassModifier))).dump()); + if (v.saveTo_fileExistsPolicy) + apply(proto::SettingKey::SaveToFileExistsPolicy, + nlohmann::json(std::string(proto::to_string(*v.saveTo_fileExistsPolicy))).dump()); + if (v.connection_preset) { + apply(proto::SettingKey::ConnectionPreset, + nlohmann::json(std::string(proto::to_string(*v.connection_preset))).dump()); + touched_connection = true; + } + if (v.downloads_duplicatePolicy) + apply(proto::SettingKey::DownloadsDuplicatePolicy, + nlohmann::json(std::string(proto::to_string(*v.downloads_duplicatePolicy))).dump()); + if (v.proxy_mode) + apply(proto::SettingKey::ProxyMode, + nlohmann::json(std::string(proto::to_string(*v.proxy_mode))).dump()); + + // connection.* governs admission (Governor/segment budget) — push it live rather than + // wait for the next restart, same trigger Scheduler::reload_config's own doc comment + // names. + if (touched_connection && actions_) actions_->apply_settings_reload(); + + if (!changed.empty()) { + hub_.publish(proto::Event::SettingsChanged, + proto::make_notification(proto::Event::SettingsChanged, + nlohmann::json{{"keys", changed}})); + } + + proto::SettingsSetResult r; + // Echo back every key the caller actually named (not the whole bag) — its stored value + // after this write, whether or not it was among the ones that changed. + for (const auto& s : kBoolSettings) + if (v.*(s.field)) r.values.*(s.field) = settings.get_bool(proto::to_string(s.key)); + for (const auto& s : kIntSettings) + if (v.*(s.field)) r.values.*(s.field) = settings.get_int(proto::to_string(s.key)); + for (const auto& s : kStrSettings) + if (v.*(s.field)) r.values.*(s.field) = settings.get_string(proto::to_string(s.key)); + for (const auto& s : kStrArrSettings) + if (v.*(s.field)) r.values.*(s.field) = settings.get_string_array(proto::to_string(s.key)); + if (v.capture_bypassModifier) + r.values.capture_bypassModifier = proto::parse_BypassModifier( + settings.get_string("capture.bypassModifier")).value_or(*v.capture_bypassModifier); + if (v.saveTo_fileExistsPolicy) + r.values.saveTo_fileExistsPolicy = + proto::parse_SettingsSaveToFileExistsPolicy(settings.get_string("saveTo.fileExistsPolicy")) + .value_or(*v.saveTo_fileExistsPolicy); + if (v.connection_preset) + r.values.connection_preset = + proto::parse_SettingsConnectionPreset(settings.get_string("connection.preset")) + .value_or(*v.connection_preset); + if (v.downloads_duplicatePolicy) + r.values.downloads_duplicatePolicy = proto::parse_SettingsDownloadsDuplicatePolicy( + settings.get_string("downloads.duplicatePolicy")).value_or(*v.downloads_duplicatePolicy); + if (v.proxy_mode) + r.values.proxy_mode = + proto::parse_SettingsProxyMode(settings.get_string("proxy.mode")).value_or(*v.proxy_mode); + r.changed = std::move(changed); + return r; } } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index a036e96..f0aa6c8 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -66,6 +66,14 @@ public: const velox::proto::DownloadProbeParams& params, std::function)> done) = 0; + + // settings.set of a connection.* key: re-read connection.maxConcurrentDownloads / + // maxActiveSegments and the per-host cap table, push the new caps to the engine and + // the governor (Scheduler::reload_config's own doc comment names this exact trigger). + // Named apply_settings_reload rather than reload_config to avoid colliding with + // sched::Scheduler's own already-public reload_config() (returns DbResult, + // consumed by main.cpp and a unit test — kept as-is rather than reshaped to fit here). + virtual void apply_settings_reload() = 0; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index c020ac2..fc68e53 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -135,6 +135,12 @@ public: bool provide_auth(const std::string& wire_id, const std::string& username, const std::string& password, bool remember) override; + // rpc::TaskActionPort::apply_settings_reload — a void-returning wrapper around the + // already-public reload_config() above (which returns DbResult, consumed by + // main.cpp and by sched_scheduler_test; kept as-is rather than changed to match the + // port, which has no caller that wants the DbError). + void apply_settings_reload() override { (void)reload_config(); } + // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a diff --git a/daemon/src/store/settings.cpp b/daemon/src/store/settings.cpp index 3b2da51..3841746 100644 --- a/daemon/src/store/settings.cpp +++ b/daemon/src/store/settings.cpp @@ -9,24 +9,62 @@ namespace velox::daemon::store { namespace { -// Built-in defaults, mirroring Settings.schema.json / ADR 0012. Only the keys the daemon -// currently reads or is likely to need before the full settings.get handler lands; the -// rest resolve through the schema's own defaults at that layer. -constexpr std::array, 14> kDefaults{{ +// Built-in defaults, one per SettingKey (Settings.schema.json / SettingKey.schema.json). +// The schema itself carries no "default" keyword anywhere — these are what settings.get +// falls back to for a key with no stored row, chosen per ADR 0012 where it speaks (the +// connection.* buffer/segment keys) and otherwise the conservative, least-surprising +// value for that key's own description. capture.monitoredExtensions defaults to the union +// of every builtin category's extensions (0001_initial.sql's seed) rather than an +// arbitrary list of its own, so the two stay in sync without a second place to edit. +constexpr std::array, 43> kDefaults{{ + {"general.launchOnLogin", "false"}, + {"general.minimizeToTray", "false"}, + {"general.showDropTarget", "true"}, + {"general.confirmOnExit", "true"}, + {"general.language", "\"system\""}, + {"general.checkForUpdates", "true"}, + {"capture.enabled", "true"}, + {"capture.monitoredExtensions", + "[\"exe\",\"msi\",\"deb\",\"rpm\",\"dmg\",\"appimage\",\"iso\",\"zip\",\"tar\",\"gz\"," + "\"xz\",\"7z\",\"mp4\",\"mkv\",\"webm\",\"avi\",\"mov\",\"flv\",\"m4v\",\"ts\",\"mp3\"," + "\"flac\",\"aac\",\"ogg\",\"opus\",\"wav\",\"m4a\",\"pdf\",\"doc\",\"docx\",\"xls\"," + "\"xlsx\",\"ppt\",\"pptx\",\"odt\",\"epub\",\"jpg\",\"jpeg\",\"png\",\"gif\",\"webp\"," + "\"svg\",\"bmp\",\"tiff\"]"}, + {"capture.monitoredMimeTypes", "[]"}, + {"capture.minSizeBytes", "0"}, + {"capture.excludedHosts", "[]"}, + {"capture.bypassModifier", "\"shift\""}, + {"capture.autoStartTypes", "[]"}, + {"saveTo.defaultDir", "\"~/Downloads\""}, + {"saveTo.tempDir", "\"\""}, + {"saveTo.allowedRoots", "[\"~/Downloads\"]"}, + {"saveTo.fileExistsPolicy", "\"ask\""}, + {"saveTo.createSubfolderPerSite", "false"}, + {"connection.preset", "\"auto\""}, {"connection.maxSegmentsPerDownload", "8"}, {"connection.bufferBytes", "1048576"}, - {"connection.maxConcurrentDownloads", "5"}, - {"connection.maxActiveSegments", "32"}, {"connection.maxTotalBufferBytes", "134217728"}, + {"connection.maxActiveSegments", "32"}, + {"connection.maxConcurrentDownloads", "5"}, {"connection.timeoutSec", "30"}, {"connection.maxRetries", "10"}, {"connection.retryBackoffSec", "5"}, - {"saveTo.defaultDir", "\"~/Downloads\""}, - {"saveTo.allowedRoots", "[\"~/Downloads\"]"}, - {"saveTo.createSubfolderPerSite", "false"}, {"downloads.speedLimitBps", "0"}, {"downloads.speedLimitEnabled", "false"}, + {"downloads.virusScanCommand", "\"\""}, + {"downloads.postDownloadCommand", "\"\""}, {"downloads.duplicatePolicy", "\"ask\""}, + {"downloads.verifyChecksums", "true"}, + {"proxy.mode", "\"system\""}, + {"proxy.host", "\"\""}, + {"proxy.port", "1"}, + {"proxy.username", "\"\""}, + {"proxy.bypassHosts", "[]"}, + {"proxy.pacUrl", "\"\""}, + {"sounds.enabled", "true"}, + {"sounds.onComplete", "\"\""}, + {"sounds.onQueueComplete", "\"\""}, + {"sounds.onError", "\"\""}, }}; } // namespace @@ -86,6 +124,19 @@ std::int64_t Settings::get_int(std::string_view key) { return 0; } +bool Settings::get_bool(std::string_view key) { + auto raw = get_raw(key); + if (raw && *raw) { + auto j = nlohmann::json::parse(**raw, nullptr, false); + if (j.is_boolean()) return j.get(); + } + if (auto d = default_for(key)) { + auto j = nlohmann::json::parse(*d, nullptr, false); + if (j.is_boolean()) return j.get(); + } + return false; +} + std::string Settings::get_string(std::string_view key) { auto raw = get_raw(key); if (raw && *raw) { diff --git a/daemon/src/store/settings.hpp b/daemon/src/store/settings.hpp index 43b9ed2..0500e2a 100644 --- a/daemon/src/store/settings.hpp +++ b/daemon/src/store/settings.hpp @@ -35,6 +35,7 @@ public: // Typed convenience over get_raw + the defaults. A malformed stored value falls back // to the default rather than throwing. std::int64_t get_int(std::string_view key); + bool get_bool(std::string_view key); std::string get_string(std::string_view key); std::vector get_string_array(std::string_view key); diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index 91803c6..b687143 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -24,3 +24,4 @@ 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) +veloxd_test(dispatcher_settings LIBS veloxd_rpc veloxd_store) diff --git a/daemon/tests/dispatcher_settings_test.cpp b/daemon/tests/dispatcher_settings_test.cpp new file mode 100644 index 0000000..4e16321 --- /dev/null +++ b/daemon/tests/dispatcher_settings_test.cpp @@ -0,0 +1,142 @@ +// settings.get / settings.set: the field <-> SettingKey <-> JSON-type mapping table in +// dispatcher.cpp, called directly (no socket) against an in-memory store. + +#include + +#include "check.hpp" +#include "rpc/dispatcher.hpp" +#include "rpc/event_hub.hpp" +#include "store/migrations.hpp" +#include "store/sqlite.hpp" +#include "velox_proto.hpp" + +using namespace velox::daemon; +namespace proto = velox::proto; + +void run() { + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + + rpc::EventHub hub; + rpc::VeloxDispatcher dispatcher(*db, hub); + + // --- get with keys=nullopt: every one of the 43 SettingKeys comes back ----------- + { + proto::SettingsGetParams p; + auto r = dispatcher.on_settings_get(p); + CHECK(r.has_value()); + if (r) { + int present = 0; + present += r->values.general_launchOnLogin.has_value(); + present += r->values.capture_enabled.has_value(); + present += r->values.saveTo_allowedRoots.has_value(); + present += r->values.connection_maxConcurrentDownloads.has_value(); + present += r->values.proxy_mode.has_value(); + present += r->values.sounds_onError.has_value(); + CHECK_EQ(present, 6); + // A default, sight-checked: connection.maxConcurrentDownloads is 5 (settings.cpp). + CHECK_EQ(r->values.connection_maxConcurrentDownloads.value_or(-1), std::int64_t{5}); + CHECK(r->values.saveTo_allowedRoots.has_value()); + if (r->values.saveTo_allowedRoots) + CHECK_EQ(r->values.saveTo_allowedRoots->size(), std::size_t{1}); + // Enum-typed default round-trips through parse_XXX correctly. + CHECK(r->values.proxy_mode == proto::SettingsProxyMode::System); + } + } + + // --- get with an explicit key list: only those fields are populated -------------- + { + proto::SettingsGetParams p; + p.keys = {proto::SettingKey::GeneralLaunchOnLogin, + proto::SettingKey::ConnectionMaxConcurrentDownloads}; + auto r = dispatcher.on_settings_get(p); + CHECK(r.has_value()); + if (r) { + CHECK(r->values.general_launchOnLogin.has_value()); + CHECK(r->values.connection_maxConcurrentDownloads.has_value()); + CHECK(!r->values.capture_enabled.has_value()); + CHECK(!r->values.proxy_mode.has_value()); + } + } + + // --- set: a valid change is persisted, reported in `changed`, and readable back --- + { + proto::SettingsSetParams p; + p.values.connection_maxConcurrentDownloads = 12; + p.values.general_launchOnLogin = true; + auto r = dispatcher.on_settings_set(p); + CHECK(r.has_value()); + if (r) { + CHECK_EQ(r->changed.size(), std::size_t{2}); + CHECK_EQ(r->values.connection_maxConcurrentDownloads.value_or(-1), std::int64_t{12}); + CHECK_EQ(r->values.general_launchOnLogin.value_or(false), true); + } + + proto::SettingsGetParams g; + g.keys = {proto::SettingKey::ConnectionMaxConcurrentDownloads}; + auto g_r = dispatcher.on_settings_get(g); + CHECK(g_r.has_value()); + if (g_r) CHECK_EQ(g_r->values.connection_maxConcurrentDownloads.value_or(-1), std::int64_t{12}); + } + + // --- set: setting the same value again reports it unchanged ---------------------- + { + proto::SettingsSetParams p; + p.values.connection_maxConcurrentDownloads = 12; + auto r = dispatcher.on_settings_set(p); + CHECK(r.has_value()); + if (r) CHECK(r->changed.empty()); + } + + // --- set: out of range is rejected, and nothing else in the same call lands ------ + { + proto::SettingsSetParams p; + p.values.connection_maxConcurrentDownloads = 999; // max 64 + p.values.general_minimizeToTray = true; // would otherwise succeed + auto r = dispatcher.on_settings_set(p); + CHECK(!r.has_value()); + if (!r) CHECK(r.error().code == proto::ErrorCode::InvalidParams); + + proto::SettingsGetParams g; + g.keys = {proto::SettingKey::GeneralMinimizeToTray}; + auto g_r = dispatcher.on_settings_get(g); + CHECK(g_r.has_value()); + // Rejected as a whole: minimizeToTray was never written even though its own value + // was valid on its own. + if (g_r) CHECK_EQ(g_r->values.general_minimizeToTray.value_or(true), false); + } + + // --- set: an enum field round-trips ------------------------------------------------ + { + proto::SettingsSetParams p; + p.values.proxy_mode = proto::SettingsProxyMode::Socks5; + auto r = dispatcher.on_settings_set(p); + CHECK(r.has_value()); + if (r) { + CHECK_EQ(r->changed.size(), std::size_t{1}); + CHECK(r->values.proxy_mode == proto::SettingsProxyMode::Socks5); + } + } + + // --- set: saveTo.defaultDir outside every allowed root is rejected with -32011 ---- + { + proto::SettingsSetParams p; + p.values.saveTo_defaultDir = "/definitely/not/an/allowed/root"; + auto r = dispatcher.on_settings_set(p); + CHECK(!r.has_value()); + if (!r) CHECK(r.error().code == proto::ErrorCode::InvalidPath); + } + + // --- set: saveTo.allowedRoots with an unresolvable entry is rejected with -32011 -- + { + proto::SettingsSetParams p; + p.values.saveTo_allowedRoots = {"/this/path/should/never/exist/anywhere"}; + auto r = dispatcher.on_settings_set(p); + CHECK(!r.has_value()); + if (!r) CHECK(r.error().code == proto::ErrorCode::InvalidPath); + } +} + +TEST_MAIN() From 6632b750992ad96d8bb88b039639fc64a2298aa1 Mon Sep 17 00:00:00 2001 From: sami Date: Sat, 12 Sep 2026 13:58:45 +0400 Subject: [PATCH 2/3] =?UTF-8?q?daemon:=20capture.offer=20for=20real=20(D7/?= =?UTF-8?q?D8)=20=E2=80=94=20rules,=20category=20resolution,=20dedupe,=20d?= =?UTF-8?q?eadline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit capture.offer applies capture.enabled/excludedHosts/monitoredExtensions/ monitoredMimeTypes/minSizeBytes from settings, then the rules table (new store::Rules + CORE's vdm::rules::match_rules/glob_match — DAEMON only converts its own stored proto::Rule JSON into CORE's plain vocabulary, per that header's own layering note), resolves the category folder (a rule's explicit categoryId/saveDir, else store::Categories::guess_by_extension — the same extension guess download.probe's suggestedCategoryId already used, now shared instead of duplicated), dedupes against active (non-terminal) tasks by exact URL, and on `take` calls add_one() — the same path download.add itself uses, so a captured download is a real, admitted, persisted task, not a special case. The 750ms deadline (CLAUDE.md §4 / AGENT-DAEMON.md build step 6) is checked cooperatively between every step via a new rpc::CaptureDataSource seam: the real implementation wraps store::Settings/Categories/Rules/Tasks; a test fake can jump its own injected clock forward to simulate "the store was slow just now" with zero real sleep. This catches the realistic failure mode (several slow steps adding up) though it cannot preempt a single pathologically stuck call mid-flight — a true preemptive guarantee would need the same async/background-thread treatment as download.probe, which isn't safe to do against the same sqlite3 connection (opened SQLITE_OPEN_NOMUTEX, explicitly not for concurrent use) without a second connection; left as a known, documented limit of this pass rather than adding that plumbing speculatively. capture.getRules returns the same settings-backed fields capture.offer itself reads, so the two can never drift. rulesVersion is a placeholder constant (1) — no persisted revision counter exists yet, and the extension already re-fetches on event.settings.changed regardless. New store/rules.{hpp,cpp}: rules table CRUD (list, and an atomic upsert+remove for rules.upsert later). store::Categories::guess_by_extension replaces a duplicate copy that used to live in sched/scheduler.cpp. store::Tasks::has_active_duplicate for the dedupe check. Verified against real veloxd + tools/testserver: a monitored-type offer answers in ~5ms and actually creates + downloads the task, correctly categorized; an unmonitored type, an excluded host, a rule-vetoed host, and a second offer for a still-active URL all answer ignore with the right reason; a bad category save dir surfaces its real -32011. New capture_offer_test covers all of the above plus the deadline itself (two cases, one per "slow" checkpoint), asserting real wall-clock time barely moves even though the fake clock jumped 2 simulated seconds. Full ctest: 54/54 (excluding the pre-existing, unrelated conformance failure noted in the previous commit). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/CMakeLists.txt | 1 + daemon/docs/deferrals.md | 4 +- daemon/src/rpc/capture_data_source.hpp | 50 +++++ daemon/src/rpc/dispatcher.cpp | 246 ++++++++++++++++++++++++- daemon/src/rpc/dispatcher.hpp | 8 + daemon/src/sched/scheduler.cpp | 26 +-- daemon/src/store/categories.cpp | 18 ++ daemon/src/store/categories.hpp | 8 + daemon/src/store/rules.cpp | 96 ++++++++++ daemon/src/store/rules.hpp | 37 ++++ daemon/src/store/tasks.cpp | 11 ++ daemon/src/store/tasks.hpp | 6 + daemon/tests/CMakeLists.txt | 1 + daemon/tests/capture_offer_test.cpp | 215 +++++++++++++++++++++ 14 files changed, 696 insertions(+), 31 deletions(-) create mode 100644 daemon/src/rpc/capture_data_source.hpp create mode 100644 daemon/src/store/rules.cpp create mode 100644 daemon/src/store/rules.hpp create mode 100644 daemon/tests/capture_offer_test.cpp diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index 4b01084..cec053a 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -37,6 +37,7 @@ add_library(veloxd_store STATIC src/store/tasks.cpp src/store/categories.cpp src/store/queues.cpp + src/store/rules.cpp src/store/segments.cpp ${_mig_hdr} ) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 014fd51..4f14e2d 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -5,8 +5,8 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | What | Where | Why deferred | Closes when | |---|---|---|---|---| -| **D7** | **`capture.offer` → `-32603`.** This is a CLAUDE.md §4 non-negotiable ("Capture fails open... the extension gives up" — but a fail-open answer still has to arrive; a hard error is not a substitute for `ignore`) and AGENT-DAEMON.md build step 6 ("`capture.offer` must answer within 750 ms, always"). Against the real daemon right now, every offer 500s, so the extension captures nothing at all — not "falls through to Firefox on a slow path," genuinely nothing, because the extension is waiting on a promise that never resolves with a decision it can act on. This was buried inside D3's stub list ("...`capture.*`") with no flag for how load-bearing it is; called out on its own row now. | `rpc/dispatcher.cpp` | rules table has no matcher yet, no dedup-against-active-tasks, no category-folder resolution | build step 6 | -| **D8** | `capture.getRules` → `-32603`. The extension mirrors this into its own capture policy on connect (docs/05 §4) so the two can never disagree about what should be intercepted; until this returns something real the extension runs with an empty/stale policy. Also buried in D3's list before this pass. | `rpc/dispatcher.cpp` | same rules-engine gap as D7 | alongside D7 | +| ~~D7~~ | **Closed — `capture.offer` is real.** Applies `capture.enabled`/`excludedHosts`/`monitoredExtensions`/`monitoredMimeTypes`/`minSizeBytes` from settings, then the rules table (`store::Rules` + CORE's `vdm::rules::match_rules`/`glob_match` — DAEMON only converts its own stored `proto::Rule` JSON into CORE's plain `vdm::rules::Rule` vocabulary, per that header's own layering note), resolves the category folder (a rule's explicit `categoryId`/`saveDir`, else `store::Categories::guess_by_extension` — the same extension-guess `download.probe`'s `suggestedCategoryId` already used, now shared instead of duplicated), dedupes against active (non-terminal) tasks by exact URL, and on `take` calls `add_one()` — the same path `download.add` itself uses — so a captured download is a real, admitted, persisted task, not a special case. The 750 ms deadline (CLAUDE.md §4 / AGENT-DAEMON.md build step 6) is checked cooperatively between every step via a new `rpc::CaptureDataSource` seam (real impl wraps `store::*`; a test fake can jump its own clock forward to simulate "the store was slow just now" with zero real sleep) — catches the realistic failure mode (several slow steps adding up) though it can't preempt one pathologically stuck single call. Verified against real `veloxd` + `tools/testserver`: a monitored-type offer answers in ~5ms and actually creates + downloads the task; an unmonitored type, an excluded host, a rule-vetoed host, and a second offer for a still-active URL all answer `ignore` with the right `reason`; a bad category save dir surfaces its real `-32011` rather than being swallowed. New `capture_offer_test` covers all of the above plus the deadline itself (two cases, one per "slow" checkpoint), asserting real wall-clock time barely moves even though the fake clock jumped 2 simulated seconds — proof the check reads the injected clock, not a disguised sleep. | `rpc/capture_data_source.hpp`, `rpc/dispatcher.{hpp,cpp}`, `store/rules.{hpp,cpp}`, `store/categories.{hpp,cpp}`, `store/tasks.{hpp,cpp}` | — | done | +| ~~D8~~ | **Closed alongside D7** — `capture.getRules` returns the same settings-backed `enabled`/`monitoredExtensions`/`monitoredMimeTypes`/`minSizeBytes`/`excludedHosts`/`bypassModifier` capture.offer itself reads, so the two can never drift. `rulesVersion` is a constant `1` — there is no persisted revision counter yet (nothing writes `rules.*` outside this process's own lifetime to need one across a restart), and the extension already re-fetches on `event.settings.changed` regardless of what this number does; noted in case a real counter becomes worth adding later. | `rpc/dispatcher.cpp` | `rulesVersion` is a placeholder constant | — | | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | | — | **D1, checked this pass, not attempted:** `libdbus-1-dev` (or `libsystemd-dev` for `sd-bus`) has no headers installed in this build environment — only the runtime `.so`s (`dpkg -l`/`apt-cache policy` confirm `libdbus-1-3` present, `libdbus-1-dev` not, "Candidate" available but not installed). A real notification-backed approver needs one of those linked into `veloxd`, which is a new build dependency for `daemon/CMakeLists.txt` (`find_package`/`pkg_check_modules`) and — since packaging manifests need to know about it too — arguably a decision to surface rather than something to reach for silently mid-session. `PairingApprover::approve()` is also still synchronous by shape (its own doc comment already says so: "the real notification-backed approver will run async and is not this shape") — swapping it for the async pattern this session built for `download.probe` (`rpc::TaskActionPort` + the server-layer deferred-reply special-case) is the right shape once there's a real implementation to justify the churn; reshaping the interface with nothing behind it yet would just be churn. Left `EnvAutoApprover` in place rather than build a fragile hand-rolled D-Bus wire client to avoid the missing headers — a broken pairing approver is worse than an honest stub. | `rpc/pairing.hpp` | missing dev headers + an undiscussed new dependency | once `libdbus-1-dev`/`libsystemd-dev` is available and the dependency is approved | | ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | diff --git a/daemon/src/rpc/capture_data_source.hpp b/daemon/src/rpc/capture_data_source.hpp new file mode 100644 index 0000000..3735e59 --- /dev/null +++ b/daemon/src/rpc/capture_data_source.hpp @@ -0,0 +1,50 @@ +#pragma once + +// The seam capture.offer's decision logic reads through, instead of touching store::* (or +// the wall clock) directly — the same reasoning as rpc::TaskActionPort: it lets a test +// substitute a fake that reports "the store took a long time just now" by advancing a +// shared fake clock, and assert that capture.offer's deadline check actually bails to +// `ignore` instead of pressing on, without a real sleep anywhere (deterministic, instant). +// +// vdm::rules::Rule (not velox::proto::Rule) on purpose: this is what +// vdm::rules::match_rules consumes directly, so the real implementation is the only place +// that ever converts the stored proto::Rule/JSON shape into CORE's plain vocabulary. + +#include +#include +#include +#include + +#include "vdm/rules/match.hpp" + +namespace velox::daemon::rpc { + +class CaptureDataSource { +public: + virtual ~CaptureDataSource() = default; + + // Read at every checkpoint in the offer's decision pipeline; a fake can advance this + // however it likes (including "jump forward 2 real seconds, instantly") to simulate a + // slow step without ever calling sleep. + virtual std::chrono::steady_clock::time_point now() = 0; + + virtual bool capture_enabled() = 0; + virtual std::vector monitored_extensions() = 0; + virtual std::vector monitored_mime_types() = 0; + virtual std::int64_t min_size_bytes() = 0; + virtual std::vector excluded_hosts() = 0; + + // Enabled rules, in priority order — ready for vdm::rules::match_rules as-is. + virtual std::vector enabled_rules() = 0; + + // "resolve the category folder": a plain extension guess (store::Categories' + // guess_by_extension) when no rule named a category explicitly. + virtual std::string guess_category_id(const std::string& filename) = 0; + virtual std::string category_save_dir(const std::string& category_id) = 0; + virtual std::string default_save_dir() = 0; + + // True if an active (non-terminal) task already targets this exact URL. + virtual bool has_active_duplicate(const std::string& url) = 0; +}; + +} // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 7490e33..c29a4b3 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -1,6 +1,7 @@ #include "rpc/dispatcher.hpp" #include +#include #include #include #include @@ -12,6 +13,7 @@ #include "fs/safepath.hpp" #include "store/categories.hpp" #include "store/queues.hpp" +#include "store/rules.hpp" #include "store/segments.hpp" #include "store/settings.hpp" #include "store/tasks.hpp" @@ -348,15 +350,251 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { return add_one(spec); } -// --- everything else : not implemented until the store and scheduler land ------------- +// --- capture.offer / capture.getRules : D7/D8 ----------------------------------------- + +namespace { + +// host[:port] out of a URL, lowercased. Duplicated from sched/scheduler.cpp's own +// file-local copy rather than shared — ten lines, no shared header either module already +// pulls in for a good reason to put it there instead. +std::string host_of(std::string_view url) { + auto scheme = url.find("://"); + std::string_view rest = scheme == std::string_view::npos ? url : url.substr(scheme + 3); + const auto at = rest.find('@'); + if (at != std::string_view::npos) rest = rest.substr(at + 1); + const auto end = rest.find_first_of("/:?#"); + std::string h(end == std::string_view::npos ? rest : rest.substr(0, end)); + std::transform(h.begin(), h.end(), h.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return h; +} + +// Bare extension, lowercased, no leading '.'; "" if `name` has none. +std::string extension_of(std::string_view name) { + const auto dot = name.find_last_of('.'); + if (dot == std::string_view::npos || dot + 1 >= name.size()) return {}; + std::string ext(name.substr(dot + 1)); + std::transform(ext.begin(), ext.end(), ext.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return ext; +} + +std::string lower(std::string s) { + std::transform(s.begin(), s.end(), s.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return s; +} + +vdm::rules::StartMode to_core_start_mode(proto::StartMode m) { + switch (m) { + case proto::StartMode::Now: return vdm::rules::StartMode::now; + case proto::StartMode::Later: return vdm::rules::StartMode::later; + case proto::StartMode::Queue: return vdm::rules::StartMode::queue; + } + return vdm::rules::StartMode::now; +} + +proto::StartMode from_core_start_mode(vdm::rules::StartMode m) { + switch (m) { + case vdm::rules::StartMode::now: return proto::StartMode::Now; + case vdm::rules::StartMode::later: return proto::StartMode::Later; + case vdm::rules::StartMode::queue: return proto::StartMode::Queue; + } + return proto::StartMode::Now; +} + +vdm::rules::Rule to_core_rule(const proto::Rule& r) { + vdm::rules::Rule out; + out.rule_id = r.ruleId; + out.enabled = r.enabled; + out.priority = r.priority; + out.match.extensions = r.match.extensions; + out.match.mime_types = r.match.mimeTypes; + out.match.host_pattern = r.match.hostPattern; + out.match.url_pattern = r.match.urlPattern; + if (r.match.minSizeBytes) out.match.min_size_bytes = static_cast(*r.match.minSizeBytes); + if (r.match.maxSizeBytes) out.match.max_size_bytes = static_cast(*r.match.maxSizeBytes); + out.action.category_id = r.action.categoryId; + out.action.save_dir = r.action.saveDir; + out.action.queue_id = r.action.queueId; + if (r.action.segments) out.action.segments = static_cast(*r.action.segments); + if (r.action.startMode) out.action.start_mode = to_core_start_mode(*r.action.startMode); + if (r.action.capture) + out.action.capture = *r.action.capture == proto::RuleActionCapture::Take + ? vdm::rules::CaptureVerdict::take + : vdm::rules::CaptureVerdict::ignore; + return out; +} + +// The real CaptureDataSource: every method is one small store read. Constructed fresh per +// call rather than held as a dispatcher member — it is stateless and db_ already outlives +// it, so there is nothing to gain from caching the object itself. +class StoreCaptureDataSource final : public CaptureDataSource { +public: + explicit StoreCaptureDataSource(store::Db& db) : db_(db) {} + + std::chrono::steady_clock::time_point now() override { + return std::chrono::steady_clock::now(); + } + + bool capture_enabled() override { return store::Settings(db_).get_bool("capture.enabled"); } + std::vector monitored_extensions() override { + return store::Settings(db_).get_string_array("capture.monitoredExtensions"); + } + std::vector monitored_mime_types() override { + return store::Settings(db_).get_string_array("capture.monitoredMimeTypes"); + } + std::int64_t min_size_bytes() override { + return store::Settings(db_).get_int("capture.minSizeBytes"); + } + std::vector excluded_hosts() override { + return store::Settings(db_).get_string_array("capture.excludedHosts"); + } + + std::vector enabled_rules() override { + std::vector out; + auto rules = store::Rules(db_).list(); + if (!rules) return out; + for (const auto& r : *rules) + if (r.enabled) out.push_back(to_core_rule(r)); + return out; + } + + std::string guess_category_id(const std::string& filename) override { + return store::Categories(db_).guess_by_extension(filename); + } + std::string category_save_dir(const std::string& category_id) override { + auto c = store::Categories(db_).get(category_id); + if (c && c->has_value()) return (*c)->saveDir; + return default_save_dir(); + } + std::string default_save_dir() override { return store::Settings(db_).get_string("saveTo.defaultDir"); } + + bool has_active_duplicate(const std::string& url) override { + return store::Tasks(db_).has_active_duplicate(url).value_or(false); + } + +private: + store::Db& db_; +}; + +} // namespace proto::HandlerResult VeloxDispatcher::on_capture_getRules(const proto::CaptureGetRulesParams&) { - return not_implemented("capture.getRules"); + store::Settings settings(db_); + proto::CaptureRules r; + r.enabled = settings.get_bool("capture.enabled"); + r.monitoredExtensions = settings.get_string_array("capture.monitoredExtensions"); + r.monitoredMimeTypes = settings.get_string_array("capture.monitoredMimeTypes"); + r.minSizeBytes = settings.get_int("capture.minSizeBytes"); + r.excludedHosts = settings.get_string_array("capture.excludedHosts"); + if (auto m = proto::parse_BypassModifier(settings.get_string("capture.bypassModifier"))) + r.bypassModifier = *m; + // No persisted revision counter exists yet; 1 is a legal starting value ("bumped on + // every change" — nothing has changed since this daemon started, so it never needs to + // bump within one run). The extension re-fetches on every event.settings.changed + // naming a capture.* key regardless of what this number does, so a restart resetting + // it to 1 costs nothing real. + r.rulesVersion = 1; + return r; } + proto::HandlerResult -VeloxDispatcher::on_capture_offer(const proto::CaptureOfferParams&) { - return not_implemented("capture.offer"); +VeloxDispatcher::on_capture_offer(const proto::CaptureOfferParams& params) { + // AGENT-DAEMON.md build step 6 / CLAUDE.md §4: answer within 750 ms, ALWAYS. 700 ms + // budget leaves 50 ms of margin for JSON serialisation and the transport write, which + // this deadline does not itself cover. Checked cooperatively between every step below + // — there is no single blocking call here to preempt (everything is a small in-process + // SQLite read), so this catches the realistic failure mode (several slow steps adding + // up) even though it cannot interrupt one pathologically stuck call mid-flight. + StoreCaptureDataSource real_source(db_); + CaptureDataSource& src = capture_source_for_test_ ? *capture_source_for_test_ : real_source; + + const auto deadline = src.now() + std::chrono::milliseconds(700); + auto ignore = [](std::optional reason) { + proto::CaptureOfferResult r; + r.action = proto::CaptureOfferResultAction::Ignore; + r.reason = reason; + return proto::HandlerResult(std::move(r)); + }; + auto deadline_exceeded = [&] { return src.now() >= deadline; }; + + if (!src.capture_enabled()) return ignore(proto::CaptureOfferResultReason::CaptureDisabled); + + const std::string host = host_of(params.url); + for (const auto& pattern : src.excluded_hosts()) + if (vdm::rules::glob_match(lower(pattern), host)) return ignore(proto::CaptureOfferResultReason::ExcludedHost); + + if (deadline_exceeded()) return ignore(std::nullopt); + + const std::string ext = extension_of(params.filename && !params.filename->empty() + ? *params.filename + : filename_from_url(params.url)); + const std::string mime = params.contentType ? lower(*params.contentType) : std::string{}; + // A `;` separates parameters from the type proper ("text/html; charset=utf-8"). + const std::string mime_type = mime.substr(0, mime.find(';')); + + const auto monitored_ext = src.monitored_extensions(); + const auto monitored_mime = src.monitored_mime_types(); + const bool ext_monitored = + !ext.empty() && std::find(monitored_ext.begin(), monitored_ext.end(), ext) != monitored_ext.end(); + const bool mime_monitored = !mime_type.empty() && std::find(monitored_mime.begin(), monitored_mime.end(), + mime_type) != monitored_mime.end(); + if (!ext_monitored && !mime_monitored) return ignore(proto::CaptureOfferResultReason::TypeNotMonitored); + + if (params.contentLength && *params.contentLength < src.min_size_bytes()) + return ignore(proto::CaptureOfferResultReason::BelowMinSize); + + if (deadline_exceeded()) return ignore(std::nullopt); + + vdm::rules::MatchInput input; + input.extension = ext; + input.mime_type = mime_type; + input.host = host; + input.url = params.url; + if (params.contentLength) input.size_bytes = static_cast(*params.contentLength); + const auto verdict = vdm::rules::match_rules(src.enabled_rules(), input); + if (verdict && verdict->capture == vdm::rules::CaptureVerdict::ignore) return ignore(proto::CaptureOfferResultReason::RuleIgnore); + + if (deadline_exceeded()) return ignore(std::nullopt); + + const std::string category_id = + (verdict && verdict->category_id) ? *verdict->category_id : src.guess_category_id(ext); + const std::string save_dir = (verdict && verdict->save_dir) ? *verdict->save_dir + : src.category_save_dir(category_id); + + if (deadline_exceeded()) return ignore(std::nullopt); + + if (src.has_active_duplicate(params.url)) return ignore(proto::CaptureOfferResultReason::Duplicate); + + if (deadline_exceeded()) return ignore(std::nullopt); + + // "resolve the category folder, ... return take/ignore": build the same DownloadSpec + // download.add itself would validate and persist — add_one() is the one place that + // turns a spec into a stored, admitted task, capture included. + proto::DownloadSpec spec; + spec.url = params.url; + spec.headers = params.headers; + spec.cookies = params.cookies; + spec.referrer = params.referrer; + spec.userAgent = params.userAgent; + if (params.filename && !params.filename->empty()) spec.filename = params.filename; + spec.saveDir = save_dir; + spec.categoryId = category_id; + if (verdict) { + spec.queueId = verdict->queue_id; + if (verdict->segments) spec.segments = static_cast(*verdict->segments); + if (verdict->start_mode) spec.startMode = from_core_start_mode(*verdict->start_mode); + } + + auto added = add_one(spec); + if (!added) return std::unexpected(added.error()); // -32011 et al. propagate as-is + + proto::CaptureOfferResult r; + r.action = proto::CaptureOfferResultAction::Take; + r.taskId = added->taskId; + return r; } proto::HandlerResult VeloxDispatcher::on_category_list(const proto::CategoryListParams&) { diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 89c1aad..1382e24 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -14,6 +14,7 @@ #include +#include "rpc/capture_data_source.hpp" #include "rpc/event_hub.hpp" #include "rpc/task_action_port.hpp" #include "store/sqlite.hpp" @@ -37,6 +38,12 @@ public: // to nudge the scheduler; unset in tests. void set_on_mutation(std::function fn) { on_mutation_ = std::move(fn); } + // Test-only seam: capture.offer normally builds its own real CaptureDataSource + // (wrapping db_) per call. A test that needs to simulate "the store is slow right + // now" (see rpc/capture_data_source.hpp) supplies one here instead; production code + // never calls this. + void set_capture_source_for_test(CaptureDataSource* src) { capture_source_for_test_ = src; } + velox::proto::HandlerResult on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override; velox::proto::HandlerResult @@ -126,6 +133,7 @@ private: EventHub& hub_; TaskActionPort* actions_; std::function on_mutation_; + CaptureDataSource* capture_source_for_test_ = nullptr; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index 61e128c..ec3705a 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -580,30 +580,6 @@ bool Scheduler::provide_auth(const std::string& wire_id, const std::string& user 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(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, @@ -635,7 +611,7 @@ void Scheduler::probe_now( 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); + r.suggestedCategoryId = store::Categories(db_).guess_by_extension(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; diff --git a/daemon/src/store/categories.cpp b/daemon/src/store/categories.cpp index d585b4b..a39e0d3 100644 --- a/daemon/src/store/categories.cpp +++ b/daemon/src/store/categories.cpp @@ -2,6 +2,8 @@ #include +#include +#include #include #include #include @@ -58,6 +60,22 @@ DbResult> Categories::get(std::string_view catego return std::optional{project_row(*st)}; } +std::string Categories::guess_by_extension(std::string_view filename_or_ext) { + std::string ext(filename_or_ext); + if (const auto dot = ext.find_last_of('.'); dot != std::string::npos) ext = ext.substr(dot + 1); + if (ext.empty()) return "general"; + std::transform(ext.begin(), ext.end(), ext.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + + auto cats = list(); + if (!cats) return "general"; + for (const auto& c : *cats) { + for (const auto& e : c.extensions) + if (e == ext) return c.categoryId; + } + return "general"; +} + DbResult Categories::upsert(proto::Category category) { // A replace keeps the existing row's builtin flag; a create is never builtin. Either // way the payload's own `builtin` is ignored — a client cannot mint or revoke it. diff --git a/daemon/src/store/categories.hpp b/daemon/src/store/categories.hpp index 1c3b57c..0e677b1 100644 --- a/daemon/src/store/categories.hpp +++ b/daemon/src/store/categories.hpp @@ -24,6 +24,14 @@ public: DbResult> list(); DbResult> get(std::string_view category_id); + // Extension match against categories.extensions — not the real rules engine (no + // host/mime/size clauses), just enough that a capture or a File Info preselect isn't + // always "general". `filename_or_ext` may be a whole filename ("movie.mp4") or a bare + // extension ("mp4", no leading dot); matched case-insensitively. "general" (this + // project's always-present default category) on no match, an empty/dotless filename, + // or a store error — this never fails outward, it just falls back. + std::string guess_by_extension(std::string_view filename_or_ext); + // "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 diff --git a/daemon/src/store/rules.cpp b/daemon/src/store/rules.cpp new file mode 100644 index 0000000..3d708b3 --- /dev/null +++ b/daemon/src/store/rules.cpp @@ -0,0 +1,96 @@ +#include "store/rules.hpp" + +#include +#include + +#include + +namespace velox::daemon::store { + +namespace proto = velox::proto; + +namespace { + +proto::Rule project_row(Stmt& st) { + proto::Rule r; + r.ruleId = st.column_text(0); + if (!st.column_is_null(1)) r.name = st.column_text(1); + r.enabled = st.column_int(2) != 0; + r.priority = st.column_int(3); + if (auto j = nlohmann::json::parse(st.column_text(4), nullptr, false); !j.is_discarded()) { + if (auto m = proto::parse(j, "match")) r.match = *m; + } + if (auto j = nlohmann::json::parse(st.column_text(5), nullptr, false); !j.is_discarded()) { + if (auto a = proto::parse(j, "action")) r.action = *a; + } + return r; +} + +std::string new_rule_id() { + std::random_device rd; + std::uniform_int_distribution d; + char buf[17]; + std::snprintf(buf, sizeof(buf), "%016llx", static_cast(d(rd))); + return std::string(buf); +} + +} // namespace + +DbResult> Rules::list() { + auto st = db_.prepare( + "SELECT rule_id, name, enabled, priority, match, action FROM rules " + "ORDER BY priority, rule_id"); + if (!st) return std::unexpected(st.error()); + + std::vector out; + for (;;) { + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + out.push_back(project_row(*st)); + } + return out; +} + +DbResult> Rules::apply(std::vector upsert, + const std::vector& remove) { + auto txn = db_.transaction([&]() -> DbResult { + for (const auto& id : remove) { + auto del = db_.prepare("DELETE FROM rules WHERE rule_id = ?1"); + if (!del) return std::unexpected(del.error()); + if (auto b = del->bind(1, std::string_view(id)); !b) return std::unexpected(b.error()); + if (auto r = del->step(); !r) return std::unexpected(r.error()); + } + for (auto& rule : upsert) { + if (rule.ruleId.empty()) rule.ruleId = new_rule_id(); + + nlohmann::json match_json = rule.match; + nlohmann::json action_json = rule.action; + + auto st = db_.prepare( + "INSERT INTO rules(rule_id, name, enabled, priority, match, action) " + "VALUES(?1,?2,?3,?4,?5,?6) " + "ON CONFLICT(rule_id) DO UPDATE SET " + "name=excluded.name, enabled=excluded.enabled, priority=excluded.priority, " + "match=excluded.match, action=excluded.action"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, std::string_view(rule.ruleId)); !b) + return std::unexpected(b.error()); + if (auto r = rule.name ? st->bind(2, std::string_view(*rule.name)) : st->bind_null(2); !r) + return std::unexpected(r.error()); + if (auto b = st->bind(3, static_cast(rule.enabled)); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(4, rule.priority); !b) return std::unexpected(b.error()); + if (auto b = st->bind(5, std::string_view(match_json.dump())); !b) + return std::unexpected(b.error()); + if (auto b = st->bind(6, std::string_view(action_json.dump())); !b) + return std::unexpected(b.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + } + return {}; + }); + if (!txn) return std::unexpected(txn.error()); + return list(); +} + +} // namespace velox::daemon::store diff --git a/daemon/src/store/rules.hpp b/daemon/src/store/rules.hpp new file mode 100644 index 0000000..bb33a47 --- /dev/null +++ b/daemon/src/store/rules.hpp @@ -0,0 +1,37 @@ +#pragma once + +// Read/write access to the `rules` table (rules.list / rules.upsert / capture.offer's own +// read path). match/action are stored as their generated-JSON text (velox::proto::RuleMatch +// / RuleAction), so no bespoke schema lives here beyond the table's own columns. + +#include +#include + +#include "store/sqlite.hpp" +#include "velox_proto.hpp" + +namespace velox::daemon::store { + +class Rules { +public: + explicit Rules(Db& db) : db_(db) {} + + // Enabled and disabled rules alike, in priority order (ties broken by rule_id) — + // rules.list's own contract ("the rules engine's table, in priority order"); capture + // offer's own caller filters to enabled ones itself, same as vdm::rules::match_rules + // already does internally. + DbResult> list(); + + // rules.upsert: "upsert carries the rules to store and remove the ruleIds to drop; + // applying both at once means a reprioritisation never leaves the table in a + // half-valid state" (the schema's own words) — one transaction. An empty ruleId in + // `upsert` generates one (create); a non-empty one replaces. Returns the full table + // after the write, in priority order. + DbResult> apply(std::vector upsert, + const std::vector& remove); + +private: + Db& db_; +}; + +} // namespace velox::daemon::store diff --git a/daemon/src/store/tasks.cpp b/daemon/src/store/tasks.cpp index 91b4f09..50b49a1 100644 --- a/daemon/src/store/tasks.cpp +++ b/daemon/src/store/tasks.cpp @@ -318,6 +318,17 @@ DbResult Tasks::count() { return (*row) ? st->column_int(0) : 0; } +DbResult Tasks::has_active_duplicate(std::string_view url) { + auto st = db_.prepare( + "SELECT 1 FROM tasks WHERE url = ?1 " + "AND state NOT IN ('complete','failed','cancelled') LIMIT 1"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, url); !b) return std::unexpected(b.error()); + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + return *row; +} + proto::TaskSummary to_summary(const TaskRow& r) { proto::TaskSummary s; s.taskId = r.task_id; diff --git a/daemon/src/store/tasks.hpp b/daemon/src/store/tasks.hpp index 4012a7b..d0c1ef7 100644 --- a/daemon/src/store/tasks.hpp +++ b/daemon/src/store/tasks.hpp @@ -79,6 +79,12 @@ public: DbResult remove(std::string_view task_id); DbResult count(); + // capture.offer's dedupe check: true if a non-terminal task already targets this exact + // URL (the same rule download.add itself does not enforce — a deliberate re-add is + // allowed there; capture is the automatic path where re-grabbing an in-flight download + // is almost always a mistake, e.g. two tabs triggering the same link). + DbResult has_active_duplicate(std::string_view url); + // Byte-counter update from an engine progress tick — cheaper than a full row rewrite, // and keeps download.list / download.get current between state transitions. DbResult update_progress(std::string_view task_id, std::int64_t downloaded_bytes, diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index b687143..561d59e 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -25,3 +25,4 @@ veloxd_test(event_hub LIBS veloxd_rpc) veloxd_test(store_categories_queues LIBS veloxd_store) veloxd_test(single_instance LIBS veloxd_rpc) veloxd_test(dispatcher_settings LIBS veloxd_rpc veloxd_store) +veloxd_test(capture_offer LIBS veloxd_rpc veloxd_store) diff --git a/daemon/tests/capture_offer_test.cpp b/daemon/tests/capture_offer_test.cpp new file mode 100644 index 0000000..162bbd3 --- /dev/null +++ b/daemon/tests/capture_offer_test.cpp @@ -0,0 +1,215 @@ +// capture.offer: excluded hosts, type/size filtering, rule matching, dedupe, take/ignore, +// and — the point of this file — the 750 ms deadline actually gets enforced when the +// store is slow, without a real sleep anywhere (a fake CaptureDataSource advances its own +// clock instead). + +#include +#include + +#include "check.hpp" +#include "rpc/capture_data_source.hpp" +#include "rpc/dispatcher.hpp" +#include "rpc/event_hub.hpp" +#include "store/migrations.hpp" +#include "store/sqlite.hpp" +#include "store/tasks.hpp" +#include "velox_proto.hpp" + +using namespace velox::daemon; +namespace proto = velox::proto; + +namespace { + +// A controllable CaptureDataSource: every accessor returns a canned value, and `now()` +// reads a clock the test (or a "slow" accessor) can jump forward instantly. No real time +// ever passes — a test that jumps 2 real seconds still runs in microseconds. +class FakeCaptureSource : public rpc::CaptureDataSource { +public: + std::chrono::steady_clock::time_point clock = std::chrono::steady_clock::now(); + bool enabled = true; + std::vector ext = {"mp4"}; + std::vector mime; + std::int64_t min_size = 0; + std::vector excluded; + std::vector rules_; + std::string category = "video"; + std::string save_dir = "~/Downloads/velox-capture-test"; + bool duplicate = false; + + // Set to jump the clock forward by this much the next time the named accessor is + // called — simulates "this particular store read took a long time." + std::chrono::milliseconds slow_on_rules{0}; + std::chrono::milliseconds slow_on_duplicate{0}; + + std::chrono::steady_clock::time_point now() override { return clock; } + bool capture_enabled() override { return enabled; } + std::vector monitored_extensions() override { return ext; } + std::vector monitored_mime_types() override { return mime; } + std::int64_t min_size_bytes() override { return min_size; } + std::vector excluded_hosts() override { return excluded; } + std::vector enabled_rules() override { + clock += slow_on_rules; + return rules_; + } + std::string guess_category_id(const std::string&) override { return category; } + std::string category_save_dir(const std::string&) override { return save_dir; } + std::string default_save_dir() override { return save_dir; } + bool has_active_duplicate(const std::string&) override { + clock += slow_on_duplicate; + return duplicate; + } +}; + +proto::CaptureOfferParams offer(std::string url) { + proto::CaptureOfferParams p; + p.url = std::move(url); + p.method = proto::CaptureOfferParamsMethod::GET; + p.tabUrl = "https://example.com/page"; + p.filename = std::string("movie.mp4"); + p.contentType = std::string("video/mp4"); + p.contentLength = 5'000'000; + return p; +} + +} // namespace + +void run() { + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + + rpc::EventHub hub; + rpc::VeloxDispatcher dispatcher(*db, hub); + + // --- capture disabled -------------------------------------------------------------- + { + FakeCaptureSource src; + src.enabled = false; + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) { + CHECK(r->action == proto::CaptureOfferResultAction::Ignore); + CHECK(r->reason == proto::CaptureOfferResultReason::CaptureDisabled); + } + } + + // --- excluded host ------------------------------------------------------------------- + { + FakeCaptureSource src; + src.excluded = {"*.excluded.example"}; + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.excluded.example/movie.mp4")); + CHECK(r.has_value()); + if (r) { + CHECK(r->action == proto::CaptureOfferResultAction::Ignore); + CHECK(r->reason == proto::CaptureOfferResultReason::ExcludedHost); + } + } + + // --- type not monitored -------------------------------------------------------------- + { + FakeCaptureSource src; + src.ext = {"iso"}; // movie.mp4 doesn't match, and mime list is empty + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) CHECK(r->reason == proto::CaptureOfferResultReason::TypeNotMonitored); + } + + // --- below minimum size ---------------------------------------------------------- + { + FakeCaptureSource src; + src.min_size = 10'000'000; // offer's contentLength is 5,000,000 + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) CHECK(r->reason == proto::CaptureOfferResultReason::BelowMinSize); + } + + // --- a rule says ignore this host -------------------------------------------------- + { + FakeCaptureSource src; + vdm::rules::Rule rule; + rule.rule_id = "r1"; + rule.enabled = true; + rule.priority = 0; + rule.match.host_pattern = "cdn.example"; + rule.action.capture = vdm::rules::CaptureVerdict::ignore; + src.rules_ = {rule}; + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) CHECK(r->reason == proto::CaptureOfferResultReason::RuleIgnore); + } + + // --- duplicate ----------------------------------------------------------------------- + { + FakeCaptureSource src; + src.duplicate = true; + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) CHECK(r->reason == proto::CaptureOfferResultReason::Duplicate); + } + + // --- take: a real task gets created, saved under the resolved category dir -------- + { + FakeCaptureSource src; + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) { + CHECK(r->action == proto::CaptureOfferResultAction::Take); + CHECK(r->taskId.has_value()); + if (r->taskId) { + store::Tasks tasks(*db); + auto row = tasks.get(*r->taskId); + CHECK(row.has_value() && row->has_value()); + if (row && *row) { + // resolve_target expands "~" and returns the canonical absolute path, + // not the literal string handed in. + CHECK((*row)->save_dir.find("velox-capture-test") != std::string::npos); + CHECK_EQ((*row)->category_id.value_or(""), std::string("video")); + } + } + } + } + + // --- the deadline: a "slow" store still gets an answer, and it's `ignore` -------- + // slow_on_rules jumps the fake clock forward 2 real seconds' worth the moment + // enabled_rules() is read (simulating a slow rules-table read); the 700 ms budget is + // long blown by the time the next checkpoint runs, so the offer must be ignored + // rather than proceeding to actually take the download. No real time passes — this + // whole test runs in microseconds. + { + FakeCaptureSource src; + src.slow_on_rules = std::chrono::milliseconds(2000); + dispatcher.set_capture_source_for_test(&src); + const auto wall_start = std::chrono::steady_clock::now(); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + const auto wall_elapsed = std::chrono::steady_clock::now() - wall_start; + CHECK(r.has_value()); + if (r) { + CHECK(r->action == proto::CaptureOfferResultAction::Ignore); + CHECK(!r->taskId.has_value()); + } + // The real wall clock barely moved — only the fake one jumped — proving the + // deadline check reads the injected clock, not a real sleep standing in for one. + CHECK(wall_elapsed < std::chrono::milliseconds(500)); + } + + // Same again, but the slow step is the dedupe check instead of rule matching — the + // deadline check has to run between every step, not just after one particular call. + { + FakeCaptureSource src; + src.slow_on_duplicate = std::chrono::milliseconds(2000); + dispatcher.set_capture_source_for_test(&src); + auto r = dispatcher.on_capture_offer(offer("https://cdn.example/movie.mp4")); + CHECK(r.has_value()); + if (r) CHECK(r->action == proto::CaptureOfferResultAction::Ignore); + } +} + +TEST_MAIN() From 4f6c0cc9d2285f151a13d4902dd09c32ec4bf0a9 Mon Sep 17 00:00:00 2001 From: sami Date: Sat, 12 Sep 2026 14:26:18 +0400 Subject: [PATCH 3/3] =?UTF-8?q?daemon:=20D3's=20remainder=20=E2=80=94=20ru?= =?UTF-8?q?les.*,=20queue.reorder,=20schedule.*,=20limiter.*,=20download.u?= =?UTF-8?q?pdate/refreshUrl?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rules.list/rules.upsert: new store/rules.{hpp,cpp} (list in priority order; apply() upserts+removes atomically, generating an id when absent, per the schema's own "never leaves the table in a half-valid state"). Found and fixed along the way: migration 0001's rules table had no column for Rule.name at all — every rules.list/.upsert call failed outright ("no such column: name"), unit tests included, since :memory: migrates through the same path. Migration 0004 adds it. queue.reorder: new store::Queues::reorder — taskIds must be an exact permutation of the queue's current membership (compared as sorted sets) or nothing is written and -32602 names the queue; a valid permutation rewrites every member's queue_position in one transaction. schedule.get/schedule.set: a thin wrapper over the queues.schedule column (already read since D3b, never independently settable). nextRunAt is deliberately left unset — computing it needs the same local-time, DST-aware window logic sched/schedule_window.hpp's window_open() only has half of; called out rather than approximated, and the field is optional. limiter.get/limiter.set: backed by the same downloads.speedLimitEnabled/ downloads.speedLimitBps settings keys D9 already wired — one bag of truth, not two. The new part is reaching the engine: EnginePort/TaskActionPort gain set_global_speed_limit(bps) (0 = unlimited, TokenBucket's own convention), wired to Engine::rate_limiter().set_global_limit(). Pushed live on every limiter.set *and* on Scheduler::reload_config() so a limit from a previous run isn't silently unlimited again after a restart. applyToRunning is accepted but has no lever to pull differently — a single shared global bucket has no "next task only" variant. Also found, not chased further: the schema's "globalBps:0 with enabled:true means 'stop everything'" is the opposite of what TokenBucket does with rate_bps==0 (unlimited) — a real discrepancy, but the schema says the GUI must not offer that combination. download.update: "moving saveDir or filename moves the file on disk in the same operation" — resolved and root-checked like download.add's destination, then the .veloxpart/.veloxpart.meta pair (or the finished file, if complete) is moved via rename, falling back to copy+remove across filesystems, only when the resolved location actually differs. categoryId/queueId(appended to the new queue's run order)/description/segments/bufferBytes/checksum apply through new store::Tasks::apply_update. download.refreshUrl: same async server-layer special-case as download.probe (a real network round trip, same 30s deadline). Re-probes, flags contentChanged only when size or validator are both known and actually differ, persists the new URL and probe result, and swaps the URL on a live engine handle via a newly-widened EnginePort::refresh_url (now takes headers too, matching DownloadHandle's real signature — the seam had silently dropped them). Found and documented, not fixed: the generated parser collapses "field absent" and "field explicitly null" to the same nullopt for every optional patch field (DownloadUpdateParamsPatch, Settings) — both schemas document "an explicit null clears the field" but neither handler can act on it because the wire distinction is already gone by the time either sees the parsed struct. A generator-level gap (PROTO's), not something to hand-route around locally. Verified against real veloxd + tools/testserver: rules create/list, limiter.set takes effect and reads back, schedule.set/get round-trips, queue.reorder against real membership (and rejects a non-permutation), download.update renames+recategorizes a task, download.refreshUrl swaps a paused task's URL and reports contentChanged correctly. Full ctest: 55/55 (excluding the pre-existing, unrelated conformance failure noted two commits back). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 9 +- daemon/src/rpc/dispatcher.cpp | 247 +++++++++++++++++- daemon/src/rpc/task_action_port.hpp | 17 ++ daemon/src/rpc/uds_server.cpp | 35 +++ daemon/src/rpc/uds_server.hpp | 3 + daemon/src/rpc/ws_server.cpp | 35 +++ daemon/src/rpc/ws_server.hpp | 1 + daemon/src/sched/engine_port.hpp | 9 +- daemon/src/sched/engine_port_core.hpp | 8 +- daemon/src/sched/fake_engine_port.hpp | 14 +- daemon/src/sched/scheduler.cpp | 83 ++++++ daemon/src/sched/scheduler.hpp | 9 + .../src/store/migrations/0004_rules_name.sql | 10 + daemon/src/store/queues.cpp | 51 ++++ daemon/src/store/queues.hpp | 12 + daemon/src/store/tasks.cpp | 82 ++++++ daemon/src/store/tasks.hpp | 28 ++ daemon/tests/CMakeLists.txt | 1 + daemon/tests/dispatcher_misc_test.cpp | 244 +++++++++++++++++ daemon/tests/sched_scheduler_test.cpp | 69 +++++ 20 files changed, 948 insertions(+), 19 deletions(-) create mode 100644 daemon/src/store/migrations/0004_rules_name.sql create mode 100644 daemon/tests/dispatcher_misc_test.cpp diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 4f14e2d..4e54da2 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -10,7 +10,14 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | | — | **D1, checked this pass, not attempted:** `libdbus-1-dev` (or `libsystemd-dev` for `sd-bus`) has no headers installed in this build environment — only the runtime `.so`s (`dpkg -l`/`apt-cache policy` confirm `libdbus-1-3` present, `libdbus-1-dev` not, "Candidate" available but not installed). A real notification-backed approver needs one of those linked into `veloxd`, which is a new build dependency for `daemon/CMakeLists.txt` (`find_package`/`pkg_check_modules`) and — since packaging manifests need to know about it too — arguably a decision to surface rather than something to reach for silently mid-session. `PairingApprover::approve()` is also still synchronous by shape (its own doc comment already says so: "the real notification-backed approver will run async and is not this shape") — swapping it for the async pattern this session built for `download.probe` (`rpc::TaskActionPort` + the server-layer deferred-reply special-case) is the right shape once there's a real implementation to justify the churn; reshaping the interface with nothing behind it yet would just be churn. Left `EnvAutoApprover` in place rather than build a fragile hand-rolled D-Bus wire client to avoid the missing headers — a broken pairing approver is worse than an honest stub. | `rpc/pairing.hpp` | missing dev headers + an undiscussed new dependency | once `libdbus-1-dev`/`libsystemd-dev` is available and the dependency is approved | | ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | -| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet | Per method, as each wires to the store/scheduler | +| D3 | Stub handlers for the rest: `grabber.*`, `media.*` | `rpc/dispatcher.cpp` | HLS/DASH grabber and media-variant support don't exist anywhere in this build yet — a bigger feature than a store-wiring pass | M4 territory, per AGENT-DAEMON.md | +| ~~D3d~~ | **Closed — `rules.list`/`rules.upsert`.** New `store/rules.{hpp,cpp}`: `list()` in priority order, `apply(upsert, remove)` in one transaction (an empty `ruleId` generates one; a reprioritisation and a removal land atomically, per the schema's own "never leaves the table in a half-valid state"). Found and fixed along the way: migration `0001`'s `rules` table had no column for `Rule.name` at all — every `rules.list`/`.upsert` call failed outright ("no such column: name") the first time either ran against a real `Db`, unit tests included, since `:memory:` migrates through the same path. Migration `0004` adds it. | `store/rules.{hpp,cpp}`, `rpc/dispatcher.cpp`, `store/migrations/0004_*.sql` | — | done | +| ~~D3e~~ | **Closed — `queue.reorder`.** New `store::Queues::reorder`: `taskIds` must be an exact permutation of the queue's current membership (compared as sorted sets) or nothing is written and `-32602` names the queue; a valid permutation rewrites every member's `queue_position` in one transaction. | `store/queues.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done | +| ~~D3f~~ | **Closed — `schedule.get`/`schedule.set`.** Thin wrapper over the `queues.schedule` column that already existed (`Queue.schema.json`'s own field, read since D3b but never independently settable). `nextRunAt` is deliberately left unset: computing it correctly needs the same local-time, DST-aware window logic `sched/schedule_window.hpp`'s `window_open()` only has half of (is-it-open-right-now, not next-transition) — real work, called out rather than approximated. The field is optional; `nullopt` is a legal answer. | `store/queues.{hpp,cpp}`, `rpc/dispatcher.cpp` | `nextRunAt` unset (documented, not silently wrong) | `nextRunAt` is its own pass | +| ~~D3g~~ | **Closed — `limiter.get`/`limiter.set`.** Backed by the same `downloads.speedLimitEnabled`/`downloads.speedLimitBps` settings keys D9 already wired (one bag of truth, not two) — the *new* part is actually reaching the engine: `EnginePort`/`TaskActionPort` gain `set_global_speed_limit(bps)` (`0` = unlimited, `vdm::rate::TokenBucket`'s own convention), wired to `vdm::Engine::rate_limiter().set_global_limit()`. Pushed live on every `limiter.set` *and* on `Scheduler::reload_config()` (so a limit from a previous run isn't silently unlimited again after a restart — nothing else re-derives it from settings the way `connection.*` already does). `applyToRunning` is accepted but has no lever to pull differently: a single shared global bucket has no "next task only" variant, so this always behaves as if it were `true` — documented in `EnginePort::set_global_speed_limit`'s own comment. Also found, not chased further: the schema's "`globalBps: 0` with `enabled: true` means 'stop everything'" is the *opposite* of what `TokenBucket` does with `rate_bps == 0` (unlimited) — a real discrepancy, but the schema itself says the GUI "must not offer" that combination, so nothing sends it in practice. | `sched/engine_port*.hpp`, `rpc/task_action_port.hpp`, `sched/scheduler.{cpp,hpp}`, `rpc/dispatcher.cpp` | the `globalBps:0` semantic clash noted above; `applyToRunning` has no real effect | flagged for whoever owns the schema/CORE conversation next | +| ~~D3h~~ | **Closed — `download.update`.** "Moving `saveDir` or `filename` moves the file on disk in the same operation" (the schema's own words): resolved and root-checked exactly like `download.add`'s destination, then the `.veloxpart`/`.veloxpart.meta` pair (or the finished file, if the task is `complete`) is moved with `std::filesystem::rename`, falling back to copy+remove across filesystems (`EXDEV`) — only if the resolved location actually differs from where the task already is. `categoryId`/`queueId`(appended to the end of the new queue's run order)/`description`/`segments`(1-32)/`bufferBytes`(64 KiB-16 MiB)/`checksum` all apply through a new `store::Tasks::apply_update`. New `store::Tasks::UpdatePatch`/`apply_update` and `Tasks::set_url` (download.refreshUrl's own need, split out since neither `apply_update` nor `set_probe_result` owns the base `url` column). | `store/tasks.{hpp,cpp}`, `rpc/dispatcher.cpp` | see the shared note below (null-clearing) | done | +| ~~D3i~~ | **Closed — `download.refreshUrl`.** Same async reasoning and the same server-layer special-case (`uds_server.cpp`/`ws_server.cpp` intercept before generic `dispatch()`, exactly like `download.probe`) — a real network round trip, same 30s `x-deadlineMs`. Re-probes the new URL, flags `contentChanged` only when size or validator are *both* known and actually differ (an unknown value on either side is never itself a mismatch — "it says so rather than silently restarting" needs a real disagreement, not an absent comparison), persists the new URL and probe result, and — if the task holds a live engine handle — swaps the URL in place via a newly-widened `EnginePort::refresh_url` (now takes headers too, matching `DownloadHandle::refresh_url`'s real signature; the seam had silently dropped them). Verified against real `veloxd` + `tools/testserver`, live-handle swap covered by a `sched_scheduler_test` case (asserts the engine got `refresh_url()`, not a fresh `start()`). | `sched/engine_port*.hpp`, `rpc/task_action_port.hpp`, `sched/scheduler.{cpp,hpp}`, `rpc/{uds_server,ws_server}.{hpp,cpp}` | — | done | +| — | **Shared note across D3h/download.update and D9/settings.set:** the generated parser collapses "field absent" and "field explicitly `null`" to the same `std::optional::nullopt` for every `optional` patch field (`DownloadUpdateParamsPatch`, `Settings`) — there is no second bit on the wire path that survives into `VeloxDispatcher`. Both schemas document "an explicit null clears the field," but neither handler can act on that distinction because the information is already gone by the time either sees the parsed struct. Not something to route around locally (would mean hand-parsing raw JSON past the generated `parse` for a handful of fields) — this is a generator-level gap, PROTO's to close (e.g. `std::optional>`, or a parallel "which fields were present" bitset). Until then: `categoryId`/`queueId`/`description`/`checksum` on `download.update`, and every nullable `Settings` key, can be *set* through these RPCs but never explicitly cleared back to null. | `contracts/` (generator), affects `rpc/dispatcher.cpp` | the wire distinction the schema documents doesn't survive to the handler | PROTO's generator | | ~~D9~~ | **Closed — `settings.get`/`settings.set`.** The field <-> `SettingKey` <-> JSON-type mapping is four pointer-to-member tables in `dispatcher.cpp` (one per C++ field type: bool, ranged int, plain string, string array) plus five enum-typed keys handled individually (`parse_XXX` already validates those); every one of the 43 `SettingKey`s now has a real default (`store::Settings::kDefaults` grew from 14 entries to 43 — `capture.monitoredExtensions`'s default is the union of every builtin category's extensions, so the two never drift apart). `settings.get` honors `keys: null` = everything. `settings.set` validates every field *before* writing any of them (numeric min/max — the schema itself carries none of this, so it's hand-checked against each key's documented range; `-32602` names the offending key, its value, and its bounds) and validates `saveTo.*` paths against `fs::resolve_target`/`canonicalize_root` (`-32011`) — `saveTo.allowedRoots` entries are checked as roots in their own right, `saveTo.defaultDir`/`.tempDir` are checked as paths resolving *inside* the (possibly, in the same call, just-updated) root list. Reports exactly the keys whose *effective* value actually changed (a `set` to the value already in effect reports `changed: []`, not the key), publishes `event.settings.changed` with that same list, and calls `TaskActionPort::apply_settings_reload()` (-> `Scheduler::reload_config()`) when any `connection.*` key took effect, live rather than waiting for a restart. Verified against real `veloxd`: all 43 keys round-trip with sane defaults, a `keys` subset filters correctly, an out-of-range value is rejected with nothing else in the same call landing, `saveTo.defaultDir` outside every allowed root is `-32011`, setting `allowedRoots` and `defaultDir` together cross-validates against the *new* roots, and `event.settings.changed` fires over a live subscription. New `dispatcher_settings_test` covers the same ground without a socket. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}`, `rpc/task_action_port.hpp`, `sched/scheduler.hpp` | — | done | | ~~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 | diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index c29a4b3..8b780ea 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -819,8 +819,120 @@ VeloxDispatcher::on_download_start(const proto::DownloadStartParams& params) { return result; } proto::HandlerResult -VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams&) { - return not_implemented("download.update"); +VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams& params) { + store::Tasks tasks(db_); + auto got = tasks.get(params.taskId); + if (!got) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "download.update: " + got.error().message}); + if (!got->has_value()) + return std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task", + nlohmann::json{{"taskId", params.taskId}}}); + const store::TaskRow& row = **got; + const auto& patch = params.patch; + + if (patch.segments && (*patch.segments < 1 || *patch.segments > 32)) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "segments must be between 1 and 32", + nlohmann::json{{"value", *patch.segments}}}); + } + if (patch.bufferBytes && (*patch.bufferBytes < 65536 || *patch.bufferBytes > 16777216)) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "bufferBytes must be between 65536 and 16777216", + nlohmann::json{{"value", *patch.bufferBytes}}}); + } + + store::Tasks::UpdatePatch update; + + // "Moving saveDir or filename moves the file on disk in the same operation" (the + // schema's own words). Resolved and root-checked exactly like download.add's own + // destination; only actually touches the filesystem if the resolved location differs + // from where the task already is. + if (patch.filename || patch.saveDir) { + store::Settings settings(db_); + std::string dir = expand_tilde(patch.saveDir.value_or(row.save_dir)); + std::string leaf = patch.filename.value_or(row.filename); + std::vector roots; + for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) + if (auto c = fs::canonicalize_root(r)) roots.push_back(*c); + + auto target = fs::resolve_target(dir, leaf, roots); + if (!target) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidPath, target.error().message, + nlohmann::json{{"path", patch.saveDir.value_or(dir)}}}); + } + + if (target->dir != row.save_dir || target->leaf != row.filename) { + namespace fsn = std::filesystem; + const std::string old_base = row.save_dir + "/" + row.filename; + const std::string new_base = target->dir + "/" + target->leaf; + + // move_one: rename, falling back to copy+remove across filesystems (EXDEV). + // Missing source (nothing to move at this path yet) is not an error. + auto move_one = [](const std::string& from, const std::string& to) -> bool { + std::error_code ec; + if (!fsn::exists(from, ec)) return true; + fsn::rename(from, to, ec); + if (!ec) return true; + ec.clear(); + fsn::copy_file(from, to, fsn::copy_options::overwrite_existing, ec); + if (ec) return false; + fsn::remove(from, ec); + return true; + }; + + const bool moved = row.state == "complete" + ? move_one(old_base, new_base) + : (move_one(old_base + ".veloxpart", new_base + ".veloxpart") && + move_one(old_base + ".veloxpart.meta", new_base + ".veloxpart.meta")); + if (!moved) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InternalError, "could not move the file to its new location", + nlohmann::json{{"path", new_base}}}); + } + update.save_dir = target->dir; + update.filename = target->leaf; + } + } + + if (patch.categoryId) update.category_id = patch.categoryId; + if (patch.queueId) { + update.queue_id = patch.queueId; + // Appended to the end of the new queue's run order. + auto page = tasks.list( + [&] { + proto::TaskFilter f; + f.queueId = *patch.queueId; + return f; + }(), + std::nullopt, 0, 10000); + std::int64_t next_pos = 0; + if (page) + for (const auto& r : page->rows) + if (r.queue_position) next_pos = std::max(next_pos, *r.queue_position + 1); + update.queue_position = next_pos; + } + if (patch.description) update.description = patch.description; + if (patch.segments) update.req_segments = patch.segments; + if (patch.bufferBytes) update.req_buffer_bytes = patch.bufferBytes; + if (patch.checksum) { + update.checksum_algo = std::string(proto::to_string(patch.checksum->algorithm)); + update.checksum_value = patch.checksum->value; + } + + if (auto r = tasks.apply_update(params.taskId, update); !r) { + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "download.update: " + r.error().message}); + } + + auto after = tasks.get(params.taskId); + if (!after || !after->has_value()) { + return std::unexpected( + proto::HandlerError{proto::ErrorCode::InternalError, "download.update: row vanished"}); + } + if (on_mutation_) on_mutation_(); + return store::to_summary(**after); } proto::HandlerResult VeloxDispatcher::on_grabber_harvest(const proto::GrabberHarvestParams&) { @@ -835,10 +947,24 @@ VeloxDispatcher::on_grabber_status(const proto::GrabberStatusParams&) { return not_implemented("grabber.status"); } proto::HandlerResult VeloxDispatcher::on_limiter_get(const proto::LimiterGetParams&) { - return not_implemented("limiter.get"); + store::Settings settings(db_); + proto::Limiter r; + r.enabled = settings.get_bool("downloads.speedLimitEnabled"); + r.globalBps = settings.get_int("downloads.speedLimitBps"); + return r; } -proto::HandlerResult VeloxDispatcher::on_limiter_set(const proto::Limiter&) { - return not_implemented("limiter.set"); +proto::HandlerResult VeloxDispatcher::on_limiter_set(const proto::Limiter& params) { + store::Settings settings(db_); + (void)settings.set_raw("downloads.speedLimitEnabled", params.enabled ? "true" : "false"); + (void)settings.set_raw("downloads.speedLimitBps", std::to_string(params.globalBps)); + + // The bucket's own convention: 0 == unlimited. `enabled: false` means "no limit" + // regardless of what globalBps happens to hold (the GUI is expected to keep it around + // for when the user re-enables it, not to reset it to 0). + if (actions_) actions_->set_global_speed_limit(params.enabled ? static_cast(params.globalBps) : 0); + + if (on_mutation_) on_mutation_(); + return params; } proto::HandlerResult VeloxDispatcher::on_media_addVariant(const proto::MediaAddVariantParams&) { @@ -860,8 +986,33 @@ VeloxDispatcher::on_queue_list(const proto::QueueListParams&) { return r; } proto::HandlerResult -VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) { - return not_implemented("queue.reorder"); +VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams& params) { + store::Queues queues(db_); + auto exists = queues.get(params.queueId); + if (!exists) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.reorder: " + exists.error().message}); + if (!exists->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such queue", + nlohmann::json{{"queueId", params.queueId}}}); + + auto applied = queues.reorder(params.queueId, params.taskIds); + if (!applied) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "queue.reorder: " + applied.error().message}); + if (!*applied) { + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, + "taskIds must be an exact permutation of the queue's current membership", + nlohmann::json{{"queueId", params.queueId}}}); + } + + auto after = queues.get(params.queueId); + proto::QueueReorderResult r; + if (after && after->has_value()) r.queue = **after; + if (on_mutation_) on_mutation_(); + return r; } proto::HandlerResult VeloxDispatcher::on_queue_start(const proto::QueueStartParams& params) { @@ -926,19 +1077,87 @@ VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams& params) { } proto::HandlerResult VeloxDispatcher::on_rules_list(const proto::RulesListParams&) { - return not_implemented("rules.list"); + store::Rules rules(db_); + auto items = rules.list(); + if (!items) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "rules.list: " + items.error().message}); + proto::RulesListResult r; + r.items = std::move(*items); + return r; } proto::HandlerResult -VeloxDispatcher::on_rules_upsert(const proto::RulesUpsertParams&) { - return not_implemented("rules.upsert"); +VeloxDispatcher::on_rules_upsert(const proto::RulesUpsertParams& params) { + store::Rules rules(db_); + auto items = rules.apply(params.upsert, params.remove.value_or(std::vector{})); + if (!items) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "rules.upsert: " + items.error().message}); + proto::RulesUpsertResult r; + r.items = std::move(*items); + return r; } proto::HandlerResult -VeloxDispatcher::on_schedule_get(const proto::ScheduleGetParams&) { - return not_implemented("schedule.get"); +VeloxDispatcher::on_schedule_get(const proto::ScheduleGetParams& params) { + store::Queues queues(db_); + proto::ScheduleGetResult r; + if (params.queueId) { + auto q = queues.get(*params.queueId); + if (!q) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "schedule.get: " + q.error().message}); + if (q->has_value()) { + proto::ScheduleGetResultItemsItem item; + item.queueId = (*q)->queueId; + item.schedule = (*q)->schedule; + r.items.push_back(std::move(item)); + } + // No such queue: an empty items list rather than an error — same "the caller asked + // about something specific and there's nothing to say about it" shape download.get + // uses -32010 for, but schedule.get's own x-errors lists only -32003, so an empty + // result (not "every schedule", just this one queue's, which doesn't exist) is the + // faithful answer within what the schema actually allows. + return r; + } + + auto all = queues.list(); + if (!all) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "schedule.get: " + all.error().message}); + for (const auto& q : *all) { + proto::ScheduleGetResultItemsItem item; + item.queueId = q.queueId; + item.schedule = q.schedule; + r.items.push_back(std::move(item)); + } + return r; } proto::HandlerResult -VeloxDispatcher::on_schedule_set(const proto::ScheduleSetParams&) { - return not_implemented("schedule.set"); +VeloxDispatcher::on_schedule_set(const proto::ScheduleSetParams& params) { + store::Queues queues(db_); + auto exists = queues.get(params.queueId); + if (!exists) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "schedule.set: " + exists.error().message}); + if (!exists->has_value()) + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidParams, "no such queue", + nlohmann::json{{"queueId", params.queueId}}}); + + auto applied = queues.set_schedule(params.queueId, params.schedule); + if (!applied) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "schedule.set: " + applied.error().message}); + + proto::ScheduleSetResult r; + r.queueId = params.queueId; + r.schedule = params.schedule; + // nextRunAt is left unset: computing it correctly needs the same local-time, + // DST-aware window logic sched/schedule_window.hpp's window_open() already has half + // of (open-right-now, not next-transition) — worth building as its own pass rather + // than an approximate guess here. The field is optional; nullopt is a legal answer. + if (on_mutation_) on_mutation_(); + return r; } proto::HandlerResult VeloxDispatcher::on_settings_get(const proto::SettingsGetParams& params) { diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index f0aa6c8..8089277 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -67,6 +67,19 @@ public: std::function)> done) = 0; + // download.refreshUrl ("IDM's 'Refresh Download Address'"): same async reasoning and + // the same server-layer special-case as probe_now — a real network round trip, up to + // the schema's own 30s x-deadlineMs. Re-probes the new URL, compares size/validator + // against what the task already has on record (contentChanged), persists the new URL + // and probe result, and — if the task holds a live engine handle — swaps its URL + // in-flight without losing progress. + virtual void refresh_url( + const std::string& wire_id, const std::string& url, + const std::optional& headers, + const std::optional>& cookies, + std::function)> + done) = 0; + // settings.set of a connection.* key: re-read connection.maxConcurrentDownloads / // maxActiveSegments and the per-host cap table, push the new caps to the engine and // the governor (Scheduler::reload_config's own doc comment names this exact trigger). @@ -74,6 +87,10 @@ public: // sched::Scheduler's own already-public reload_config() (returns DbResult, // consumed by main.cpp and a unit test — kept as-is rather than reshaped to fit here). virtual void apply_settings_reload() = 0; + + // limiter.set: pushed straight to the engine's shared global token bucket. 0 means + // unlimited (the bucket's own convention). + virtual void set_global_speed_limit(std::uint64_t bps) = 0; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp index aaf0ac4..2362ad5 100644 --- a/daemon/src/rpc/uds_server.cpp +++ b/daemon/src/rpc/uds_server.cpp @@ -207,6 +207,10 @@ void UdsServer::handle_line(Conn& c, const std::string& line) { handle_download_probe(c, req); return; } + if (method == "download.refreshUrl") { + handle_download_refreshUrl(c, req); + return; + } // Everything else: the generated router. It returns a null json for a notification // that needs no reply. @@ -244,6 +248,37 @@ void UdsServer::handle_download_probe(Conn& c, const json& request) { }); } +void UdsServer::handle_download_refreshUrl(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.refreshUrl")); + return; + } + + const int fd = c.fd; + actions_->refresh_url( + parsed->taskId, parsed->url, parsed->headers, parsed->cookies, + [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + queue_reply(*it->second, proto::make_result(id, *r)); + } else { + queue_reply(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool UdsServer::handle_session_method(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/uds_server.hpp b/daemon/src/rpc/uds_server.hpp index d76f7a8..a196c0a 100644 --- a/daemon/src/rpc/uds_server.hpp +++ b/daemon/src/rpc/uds_server.hpp @@ -76,6 +76,9 @@ private: // 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); + // Same reasoning as handle_download_probe — a real network round trip, same 30s + // x-deadlineMs. + void handle_download_refreshUrl(Conn& c, const nlohmann::json& request); void queue_reply(Conn& c, const nlohmann::json& reply); void flush(Conn& c); diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp index 875bee5..25c46b2 100644 --- a/daemon/src/rpc/ws_server.cpp +++ b/daemon/src/rpc/ws_server.cpp @@ -267,6 +267,10 @@ void WsServer::handle_rpc(Conn& c, const std::string& text) { handle_download_probe(c, req); return; } + if (method == "download.refreshUrl") { + handle_download_refreshUrl(c, req); + return; + } json reply = proto::dispatch(dispatcher_, proto::Transport::Ws, req); if (!reply.is_null()) send_text(c, reply); @@ -302,6 +306,37 @@ void WsServer::handle_download_probe(Conn& c, const json& request) { }); } +void WsServer::handle_download_refreshUrl(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + send_text(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + send_text(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.refreshUrl")); + return; + } + + const int fd = c.fd; + actions_->refresh_url( + parsed->taskId, parsed->url, parsed->headers, parsed->cookies, + [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + send_text(*it->second, proto::make_result(id, *r)); + } else { + send_text(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/ws_server.hpp b/daemon/src/rpc/ws_server.hpp index bc71864..ce4d78e 100644 --- a/daemon/src/rpc/ws_server.hpp +++ b/daemon/src/rpc/ws_server.hpp @@ -81,6 +81,7 @@ private: // 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 handle_download_refreshUrl(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); diff --git a/daemon/src/sched/engine_port.hpp b/daemon/src/sched/engine_port.hpp index b487eb7..7c7ec87 100644 --- a/daemon/src/sched/engine_port.hpp +++ b/daemon/src/sched/engine_port.hpp @@ -46,7 +46,8 @@ public: virtual void provide_auth(vdm::TaskId, const std::string& username, const std::string& password, bool remember) = 0; virtual void decide(vdm::TaskId, vdm::task::Decision) = 0; - virtual void refresh_url(vdm::TaskId, const std::string& url) = 0; + virtual void refresh_url(vdm::TaskId, const std::string& url, + const std::vector& headers = {}) = 0; // The daemon is done with this task (it went terminal). Drop the handle. Idempotent. virtual void release(vdm::TaskId) = 0; @@ -59,6 +60,12 @@ public: virtual void set_task_order(const std::vector& order) = 0; virtual void set_max_active_segments(std::uint32_t n) = 0; virtual void set_host_segment_cap(const std::string& host, std::uint32_t cap) = 0; + + // limiter.set: 0 means unlimited (vdm::rate::TokenBucket's own convention), applied + // across every active transfer immediately — there is no "next task only" variant for + // a single shared global bucket, so `applyToRunning` on the wire has nothing to select + // between; it is accepted for schema compliance and always behaves as if true. + virtual void set_global_speed_limit(std::uint64_t bps) = 0; }; } // namespace velox::daemon::sched diff --git a/daemon/src/sched/engine_port_core.hpp b/daemon/src/sched/engine_port_core.hpp index 7c6fbe4..7084995 100644 --- a/daemon/src/sched/engine_port_core.hpp +++ b/daemon/src/sched/engine_port_core.hpp @@ -45,8 +45,9 @@ public: void decide(vdm::TaskId id, vdm::task::Decision d) override { if (auto* h = find(id)) h->decide(d); } - void refresh_url(vdm::TaskId id, const std::string& url) override { - if (auto* h = find(id)) h->refresh_url(url); + void refresh_url(vdm::TaskId id, const std::string& url, + const std::vector& headers) override { + if (auto* h = find(id)) h->refresh_url(url, headers); } void release(vdm::TaskId id) override { handles_.erase(id); } std::optional progress(vdm::TaskId id) const override { @@ -64,6 +65,9 @@ public: void set_host_segment_cap(const std::string& host, std::uint32_t cap) override { engine_.segment_budget().set_host_segment_cap(host, cap); } + void set_global_speed_limit(std::uint64_t bps) override { + engine_.rate_limiter().set_global_limit(bps); + } private: vdm::task::DownloadHandle* find(vdm::TaskId id) { diff --git a/daemon/src/sched/fake_engine_port.hpp b/daemon/src/sched/fake_engine_port.hpp index 1e88d81..8027b2a 100644 --- a/daemon/src/sched/fake_engine_port.hpp +++ b/daemon/src/sched/fake_engine_port.hpp @@ -61,7 +61,16 @@ public: void cancel(vdm::TaskId id, bool discard) override { cancelled.emplace_back(id, discard); } void provide_auth(vdm::TaskId, const std::string&, const std::string&, bool) override {} void decide(vdm::TaskId, vdm::task::Decision) override {} - void refresh_url(vdm::TaskId, const std::string&) override {} + void refresh_url(vdm::TaskId id, const std::string& url, + const std::vector& headers) override { + refreshed_urls.emplace_back(id, url, headers); + } + struct RefreshCall { + vdm::TaskId id; + std::string url; + std::vector headers; + }; + std::vector refreshed_urls; void release(vdm::TaskId id) override { released.push_back(id); } std::optional progress(vdm::TaskId id) const override { auto it = fake_progress.find(id.value); @@ -75,6 +84,9 @@ public: void set_host_segment_cap(const std::string& h, std::uint32_t c) override { host_caps.emplace_back(h, c); } + void set_global_speed_limit(std::uint64_t bps) override { global_speed_limits.push_back(bps); } + + std::vector global_speed_limits; const std::vector& last_order() const { return orders.back(); } diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index ec3705a..d13b772 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -183,6 +183,16 @@ store::DbResult Scheduler::reload_config() { governor_.set_config(cfg); engine_.set_max_active_segments( static_cast(std::max(cfg.max_active_segments, 1))); + + // The global speed limit persists across a restart the same as any other setting, but + // (unlike connection.* above) nothing re-derives it into engine state on its own — + // limiter.set is the only other place that calls set_global_speed_limit, and that only + // fires on an explicit RPC in a running daemon. Push it here too so a limit set in a + // previous run is not silently unlimited again after a restart. + const bool limit_enabled = settings.get_bool("downloads.speedLimitEnabled"); + const std::int64_t limit_bps = settings.get_int("downloads.speedLimitBps"); + engine_.set_global_speed_limit(limit_enabled ? static_cast(std::max(limit_bps, 0)) + : 0); return {}; } @@ -634,4 +644,77 @@ void Scheduler::probe_now( }); } +void Scheduler::refresh_url( + const std::string& wire_id, const std::string& url, + const std::optional& headers, + const std::optional>& cookies, + std::function)> done) { + auto got = store::Tasks(db_).get(wire_id); + if (!got || !got->has_value()) { + done(std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task", + nlohmann::json{{"taskId", wire_id}}})); + return; + } + const store::TaskRow row = **got; // copied: read again by the async callback below + + vdm::net::ProbeRequest req; + req.url = url; + if (headers) + for (const auto& [k, v] : *headers) req.headers.push_back({k, v}); + if (cookies) + for (const auto& c : *cookies) req.cookies.push_back({c.name, c.value}); + + engine_.probe(req, [this, wire_id, row, url, headers, done](vdm::Result pr) { + deps_.post_to_loop([this, wire_id, row, url, headers, 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; + } + + // "if they do not [match], it says so rather than silently restarting" (the + // schema's own words) — comparison only fires when both sides actually have a + // value; an unknown size/validator on either end is not itself a mismatch. + bool content_changed = false; + if (row.size_bytes && pr->total_size && + *row.size_bytes != static_cast(*pr->total_size)) + content_changed = true; + if (row.etag && !row.etag->empty() && !pr->etag.empty() && *row.etag != pr->etag) + content_changed = true; + if (row.last_modified && !row.last_modified->empty() && !pr->last_modified.empty() && + *row.last_modified != pr->last_modified) + content_changed = true; + + store::Tasks tasks(db_); + store::Tasks::ProbeFields fields; + if (pr->total_size) fields.size_bytes = static_cast(*pr->total_size); + fields.resumable = pr->resumable; + if (!pr->etag.empty()) fields.etag = pr->etag; + if (!pr->last_modified.empty()) fields.last_modified = pr->last_modified; + if (!pr->mime.empty()) fields.content_type = pr->mime; + const std::string effective = pr->effective_url.empty() ? url : pr->effective_url; + fields.effective_url = effective; + (void)tasks.set_probe_result(wire_id, fields); + (void)tasks.set_url(wire_id, url); + + if (auto eid = engine_id_of(wire_id)) { + std::vector hdrs; + if (headers) + for (const auto& [k, v] : *headers) hdrs.push_back({k, v}); + engine_.refresh_url(*eid, url, hdrs); + } + + proto::DownloadRefreshUrlResult r; + r.ok = true; + r.resumable = pr->resumable; + r.contentChanged = content_changed; + if (pr->total_size) r.sizeBytes = static_cast(*pr->total_size); + r.effectiveUrl = effective; + done(r); + }); + }); +} + } // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index fc68e53..afa4afd 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -141,6 +141,8 @@ public: // port, which has no caller that wants the DbError). void apply_settings_reload() override { (void)reload_config(); } + void set_global_speed_limit(std::uint64_t bps) override { engine_.set_global_speed_limit(bps); } + // 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 @@ -153,6 +155,13 @@ public: std::function)> done) override; + void refresh_url( + const std::string& wire_id, const std::string& url, + const std::optional& headers, + const std::optional>& cookies, + std::function)> + done) override; + // Diagnostics / tests. std::optional wire_id_of(vdm::TaskId id) const; std::optional engine_id_of(const std::string& wire_id) const; diff --git a/daemon/src/store/migrations/0004_rules_name.sql b/daemon/src/store/migrations/0004_rules_name.sql new file mode 100644 index 0000000..759e253 --- /dev/null +++ b/daemon/src/store/migrations/0004_rules_name.sql @@ -0,0 +1,10 @@ +-- Migration 0004 — rules gets a name column. +-- +-- 0001's rules table had no column for Rule.name (Rule.schema.json's own optional, +-- maxLength-64 label field) — store/rules.cpp discovered this the hard way building +-- rules.list/rules.upsert: every rules.list call failed outright ("no such column: +-- name") because the SELECT it needs to project onto proto::Rule names a column that was +-- never there. A plain ALTER TABLE ADD COLUMN suffices here (no CHECK constraint to +-- rebuild around, unlike 0002/0003). + +ALTER TABLE rules ADD COLUMN name TEXT; diff --git a/daemon/src/store/queues.cpp b/daemon/src/store/queues.cpp index 8db8771..ffe8769 100644 --- a/daemon/src/store/queues.cpp +++ b/daemon/src/store/queues.cpp @@ -2,6 +2,7 @@ #include +#include #include #include #include @@ -85,6 +86,56 @@ DbResult Queues::set_state(std::string_view queue_id, std::string_view sta return sqlite3_changes(db_.raw()) > 0; } +DbResult Queues::set_schedule(std::string_view queue_id, + const std::optional& schedule) { + auto st = db_.prepare("UPDATE queues SET schedule = ?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 r = schedule ? st->bind(2, std::string_view(nlohmann::json(*schedule).dump())) + : st->bind_null(2); + !r) + return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + +DbResult Queues::reorder(std::string_view queue_id, const std::vector& task_ids) { + std::vector current; + { + auto st = db_.prepare( + "SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position"); + if (!st) return std::unexpected(st.error()); + if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error()); + for (;;) { + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + current.push_back(st->column_text(0)); + } + } + + // Exact permutation: same size, same members, order aside. + std::vector a = current, b = task_ids; + std::sort(a.begin(), a.end()); + std::sort(b.begin(), b.end()); + if (a != b) return false; + + auto txn = db_.transaction([&]() -> DbResult { + for (std::size_t i = 0; i < task_ids.size(); ++i) { + auto st = db_.prepare("UPDATE tasks SET queue_position = ?2 WHERE task_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto bd = st->bind(1, std::string_view(task_ids[i])); !bd) + return std::unexpected(bd.error()); + if (auto bd = st->bind(2, static_cast(i)); !bd) + return std::unexpected(bd.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + } + return {}; + }); + if (!txn) return std::unexpected(txn.error()); + return true; +} + DbResult Queues::upsert(proto::Queue queue) { if (queue.queueId.empty()) { std::random_device rd; diff --git a/daemon/src/store/queues.hpp b/daemon/src/store/queues.hpp index a284aea..6b36599 100644 --- a/daemon/src/store/queues.hpp +++ b/daemon/src/store/queues.hpp @@ -27,6 +27,18 @@ public: // id doesn't exist. DbResult set_state(std::string_view queue_id, std::string_view state); + // schedule.set: nullopt clears it (NULL = manual control, per the schema's own + // words). false if the queue doesn't exist. + DbResult set_schedule(std::string_view queue_id, + const std::optional& schedule); + + // queue.reorder: `task_ids` must be an exact permutation of the queue's current + // membership (the schema's own words — "anything else is -32602 rather than a + // partial reorder, so a stale drag from an out-of-date view cannot quietly reshuffle + // the queue"). false (no write at all) if it isn't; true and queue_position rewritten + // to match `task_ids`'s order if it is. + DbResult reorder(std::string_view queue_id, const std::vector& task_ids); + // "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 diff --git a/daemon/src/store/tasks.cpp b/daemon/src/store/tasks.cpp index 50b49a1..480c91a 100644 --- a/daemon/src/store/tasks.cpp +++ b/daemon/src/store/tasks.cpp @@ -318,6 +318,88 @@ DbResult Tasks::count() { return (*row) ? st->column_int(0) : 0; } +DbResult Tasks::apply_update(std::string_view task_id, const UpdatePatch& patch) { + // One UPDATE per present field: simplest thing that's obviously correct for a + // single-row edit with ~9 independent optional fields, and it means a field the + // caller didn't touch is never rewritten with its own unchanged value (matters for + // no-op-detection callers, though download.update doesn't currently need that). + bool touched_any = false; + auto run = [&](const char* sql, auto&& binder) -> DbResult { + auto st = db_.prepare(sql); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = binder(*st); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + touched_any = true; + return {}; + }; + + if (patch.save_dir && patch.filename) { + if (auto r = run("UPDATE tasks SET save_dir = ?2, filename = ?3 WHERE task_id = ?1", + [&](Stmt& s) { + if (auto b = s.bind(2, std::string_view(*patch.save_dir)); !b) return b; + return s.bind(3, std::string_view(*patch.filename)); + }); + !r) + return std::unexpected(r.error()); + } + if (patch.category_id) { + if (auto r = run("UPDATE tasks SET category_id = ?2 WHERE task_id = ?1", + [&](Stmt& s) { return s.bind(2, std::string_view(*patch.category_id)); }); + !r) + return std::unexpected(r.error()); + } + if (patch.queue_id) { + if (auto r = run("UPDATE tasks SET queue_id = ?2, queue_position = ?3 WHERE task_id = ?1", + [&](Stmt& s) { + if (auto b = s.bind(2, std::string_view(*patch.queue_id)); !b) return b; + return patch.queue_position ? s.bind(3, *patch.queue_position) + : s.bind_null(3); + }); + !r) + return std::unexpected(r.error()); + } + if (patch.description) { + if (auto r = run("UPDATE tasks SET description = ?2 WHERE task_id = ?1", + [&](Stmt& s) { return s.bind(2, std::string_view(*patch.description)); }); + !r) + return std::unexpected(r.error()); + } + if (patch.req_segments) { + if (auto r = run("UPDATE tasks SET req_segments = ?2 WHERE task_id = ?1", + [&](Stmt& s) { return s.bind(2, *patch.req_segments); }); + !r) + return std::unexpected(r.error()); + } + if (patch.req_buffer_bytes) { + if (auto r = run("UPDATE tasks SET req_buffer_bytes = ?2 WHERE task_id = ?1", + [&](Stmt& s) { return s.bind(2, *patch.req_buffer_bytes); }); + !r) + return std::unexpected(r.error()); + } + if (patch.checksum_algo && patch.checksum_value) { + if (auto r = run( + "UPDATE tasks SET checksum_algo = ?2, checksum_value = ?3 WHERE task_id = ?1", + [&](Stmt& s) { + if (auto b = s.bind(2, std::string_view(*patch.checksum_algo)); !b) return b; + return s.bind(3, std::string_view(*patch.checksum_value)); + }); + !r) + return std::unexpected(r.error()); + } + + return touched_any; +} + +DbResult Tasks::set_url(std::string_view task_id, std::string_view url) { + auto st = db_.prepare("UPDATE tasks SET url = ?2 WHERE task_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = st->bind(2, url); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + DbResult Tasks::has_active_duplicate(std::string_view url) { auto st = db_.prepare( "SELECT 1 FROM tasks WHERE url = ?1 " diff --git a/daemon/src/store/tasks.hpp b/daemon/src/store/tasks.hpp index d0c1ef7..5c71696 100644 --- a/daemon/src/store/tasks.hpp +++ b/daemon/src/store/tasks.hpp @@ -85,6 +85,34 @@ public: // is almost always a mistake, e.g. two tabs triggering the same link). DbResult has_active_duplicate(std::string_view url); + // download.update's patch, already resolved by the caller (new save_dir/filename + // canonicalized and root-checked, any file already moved on disk — this only writes + // the row). Every field is applied when present; queue_position is written alongside + // queue_id (nullopt leaves the existing position alone — the caller decides what + // "moved into a queue" should set it to). Note: the generated parser collapses "field + // absent" and "field explicitly null" to the same nullopt (DownloadUpdateParamsPatch + // has no way to tell them apart on the wire as generated), so this — like the RPC + // layer above it — can only ever set category_id/queue_id/description/checksum, never + // clear them back to NULL through this call. + struct UpdatePatch { + std::optional save_dir; + std::optional filename; + std::optional category_id; + std::optional queue_id; + std::optional queue_position; + std::optional description; + std::optional req_segments; + std::optional req_buffer_bytes; + std::optional checksum_algo; + std::optional checksum_value; + }; + DbResult apply_update(std::string_view task_id, const UpdatePatch& patch); + + // download.refreshUrl: point the task at a freshly-issued URL. Separate from + // apply_update/set_probe_result since neither owns the base `url` column — refreshUrl + // is the one caller that changes it after creation. + DbResult set_url(std::string_view task_id, std::string_view url); + // Byte-counter update from an engine progress tick — cheaper than a full row rewrite, // and keeps download.list / download.get current between state transitions. DbResult update_progress(std::string_view task_id, std::int64_t downloaded_bytes, diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index 561d59e..17ece28 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -26,3 +26,4 @@ veloxd_test(store_categories_queues LIBS veloxd_store) veloxd_test(single_instance LIBS veloxd_rpc) veloxd_test(dispatcher_settings LIBS veloxd_rpc veloxd_store) veloxd_test(capture_offer LIBS veloxd_rpc veloxd_store) +veloxd_test(dispatcher_misc LIBS veloxd_rpc veloxd_store) diff --git a/daemon/tests/dispatcher_misc_test.cpp b/daemon/tests/dispatcher_misc_test.cpp new file mode 100644 index 0000000..afbb7a3 --- /dev/null +++ b/daemon/tests/dispatcher_misc_test.cpp @@ -0,0 +1,244 @@ +// rules.list/upsert, queue.reorder, schedule.get/set, limiter.get/set, download.update: +// the rest of D3, called directly against an in-memory store (no socket). + +#include + +#include "check.hpp" +#include "rpc/dispatcher.hpp" +#include "rpc/event_hub.hpp" +#include "store/migrations.hpp" +#include "store/queues.hpp" +#include "store/sqlite.hpp" +#include "store/tasks.hpp" +#include "velox_proto.hpp" + +using namespace velox::daemon; +namespace proto = velox::proto; + +namespace { + +store::TaskRow task(std::string id, std::optional queue = std::nullopt, + std::int64_t pos = 0) { + store::TaskRow r; + r.task_id = std::move(id); + r.url = "https://cdn.example/" + r.task_id; + r.save_dir = "/tmp"; + r.filename = r.task_id + ".bin"; + r.state = "queued"; + r.created_at = "2026-09-12T00:00:00Z"; + r.queue_id = std::move(queue); + if (r.queue_id) r.queue_position = pos; + return r; +} + +} // namespace + +void run() { + auto db = store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(store::migrate_to_head(*db).has_value()); + + rpc::EventHub hub; + rpc::VeloxDispatcher dispatcher(*db, hub); + + // --- rules.list starts empty; rules.upsert creates, replaces, and removes -------- + { + auto listed = dispatcher.on_rules_list({}); + CHECK(listed.has_value()); + if (listed) CHECK(listed->items.empty()); + + proto::RulesUpsertParams up; + proto::Rule r; + r.name = "ISOs to a queue"; + r.enabled = true; + r.priority = 5; + r.match.extensions = std::vector{"iso"}; + r.action.categoryId = "programs"; + up.upsert = {r}; + auto created = dispatcher.on_rules_upsert(up); + CHECK(created.has_value()); + if (created) { + CHECK_EQ(created->items.size(), std::size_t{1}); + CHECK(!created->items[0].ruleId.empty()); + } + + // Replace: same ruleId, new priority. + if (!created || created->items.empty()) return; + std::string rule_id = created->items[0].ruleId; + proto::RulesUpsertParams replace; + proto::Rule r2 = created->items[0]; + r2.priority = 1; + replace.upsert = {r2}; + auto replaced = dispatcher.on_rules_upsert(replace); + CHECK(replaced.has_value()); + if (replaced) { + CHECK_EQ(replaced->items.size(), std::size_t{1}); + CHECK_EQ(replaced->items[0].priority, std::int64_t{1}); + } + + // Remove. + proto::RulesUpsertParams rm; + rm.remove = {rule_id}; + auto removed = dispatcher.on_rules_upsert(rm); + CHECK(removed.has_value()); + if (removed) CHECK(removed->items.empty()); + } + + // --- queue.reorder: a real permutation applies; a non-permutation is -32602 ------ + { + store::Tasks tasks(*db); + CHECK(tasks.insert(task("q0", "main", 0)).has_value()); + CHECK(tasks.insert(task("q1", "main", 1)).has_value()); + CHECK(tasks.insert(task("q2", "main", 2)).has_value()); + + proto::QueueReorderParams p; + p.queueId = "main"; + p.taskIds = {"q2", "q0", "q1"}; + auto r = dispatcher.on_queue_reorder(p); + CHECK(r.has_value()); + if (r) { + CHECK(r->queue.taskIds.has_value()); + if (r->queue.taskIds) { + const auto& ids = *r->queue.taskIds; + CHECK_EQ(ids.size(), std::size_t{3}); + if (ids.size() == 3) { + CHECK_EQ(ids[0], std::string("q2")); + CHECK_EQ(ids[1], std::string("q0")); + CHECK_EQ(ids[2], std::string("q1")); + } + } + } + + proto::QueueReorderParams bad; + bad.queueId = "main"; + bad.taskIds = {"q0", "q1"}; // missing q2: not a permutation + auto bad_r = dispatcher.on_queue_reorder(bad); + CHECK(!bad_r.has_value()); + if (!bad_r) CHECK(bad_r.error().code == proto::ErrorCode::InvalidParams); + + proto::QueueReorderParams unknown_queue; + unknown_queue.queueId = "does-not-exist"; + unknown_queue.taskIds = {}; + auto unknown_r = dispatcher.on_queue_reorder(unknown_queue); + CHECK(!unknown_r.has_value()); + + tasks.remove("q0"); + tasks.remove("q1"); + tasks.remove("q2"); + } + + // --- schedule.get / schedule.set --------------------------------------------------- + { + proto::ScheduleGetParams g; + g.queueId = std::string("main"); + auto before = dispatcher.on_schedule_get(g); + CHECK(before.has_value()); + if (before) { + CHECK_EQ(before->items.size(), std::size_t{1}); + if (!before->items.empty()) CHECK(!before->items[0].schedule.has_value()); + } + + proto::ScheduleSetParams set; + set.queueId = "main"; + proto::Schedule sched; + sched.enabled = true; + sched.mode = proto::ScheduleMode::Periodic; + sched.startTime = std::string("22:00"); + sched.stopTime = std::string("06:00"); + set.schedule = sched; + auto set_r = dispatcher.on_schedule_set(set); + CHECK(set_r.has_value()); + if (set_r) { + CHECK(set_r->schedule.has_value()); + CHECK(!set_r->nextRunAt.has_value()); // documented gap, not computed + } + + auto after = dispatcher.on_schedule_get(g); + CHECK(after.has_value()); + if (after && !after->items.empty()) + CHECK(after->items[0].schedule.has_value()); + + // Clearing (the wire-level "explicit null" is unreachable through the generated + // parser's collapsing of absent/null — see UpdatePatch's own comment; this test + // only proves set-a-value works, not clear). + + proto::ScheduleGetParams all; + auto all_r = dispatcher.on_schedule_get(all); + CHECK(all_r.has_value()); + if (all_r) CHECK(!all_r->items.empty()); + + proto::ScheduleGetParams missing; + missing.queueId = std::string("does-not-exist"); + auto missing_r = dispatcher.on_schedule_get(missing); + CHECK(missing_r.has_value()); + if (missing_r) CHECK(missing_r->items.empty()); + } + + // --- limiter.get / limiter.set ----------------------------------------------------- + { + auto before = dispatcher.on_limiter_get({}); + CHECK(before.has_value()); + if (before) { + CHECK(!before->enabled); + CHECK_EQ(before->globalBps, std::int64_t{0}); + } + + proto::Limiter set; + set.enabled = true; + set.globalBps = 2'000'000; + auto set_r = dispatcher.on_limiter_set(set); + CHECK(set_r.has_value()); + if (set_r) { + CHECK(set_r->enabled); + CHECK_EQ(set_r->globalBps, std::int64_t{2'000'000}); + } + + auto after = dispatcher.on_limiter_get({}); + CHECK(after.has_value()); + if (after) { + CHECK(after->enabled); + CHECK_EQ(after->globalBps, std::int64_t{2'000'000}); + } + } + + // --- download.update: metadata fields, range validation, not-found --------------- + { + store::Tasks tasks(*db); + CHECK(tasks.insert(task("u0")).has_value()); + + proto::DownloadUpdateParams p; + p.taskId = "u0"; + p.patch.description = std::string("a note"); + p.patch.segments = 4; + auto r = dispatcher.on_download_update(p); + CHECK(r.has_value()); + if (r) { + CHECK_EQ(r->taskId, std::string("u0")); + } + auto row = tasks.get("u0"); + CHECK(row.has_value() && row->has_value()); + if (row && *row) { + CHECK_EQ((*row)->description.value_or(""), std::string("a note")); + CHECK_EQ((*row)->req_segments.value_or(-1), std::int64_t{4}); + } + + proto::DownloadUpdateParams bad_range; + bad_range.taskId = "u0"; + bad_range.patch.segments = 999; + auto bad_r = dispatcher.on_download_update(bad_range); + CHECK(!bad_r.has_value()); + if (!bad_r) CHECK(bad_r.error().code == proto::ErrorCode::InvalidParams); + + proto::DownloadUpdateParams missing; + missing.taskId = "does-not-exist"; + missing.patch.description = std::string("x"); + auto missing_r = dispatcher.on_download_update(missing); + CHECK(!missing_r.has_value()); + if (!missing_r) CHECK(missing_r.error().code == proto::ErrorCode::TaskNotFound); + + tasks.remove("u0"); + } +} + +TEST_MAIN() diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp index 8f90977..b515748 100644 --- a/daemon/tests/sched_scheduler_test.cpp +++ b/daemon/tests/sched_scheduler_test.cpp @@ -406,6 +406,75 @@ void run() { if (got2 && !*got2) CHECK(got2->error().code == velox::proto::ErrorCode::ProbeFailed); } + + // --- refresh_url: content unchanged, content changed, and a live-handle swap ----- + { + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + store::TaskRow r; + r.task_id = "ru0"; + r.url = "https://cdn.example/old-signed-url"; + r.save_dir = "/tmp"; + r.filename = "ru0.bin"; + r.state = "queued"; + r.created_at = "2026-09-12T00:00:00Z"; + r.size_bytes = 1000; + r.etag = "\"same\""; + CHECK(tasks.insert(r).has_value()); + CHECK(sched.tick().has_value()); + CHECK_EQ(engine.starts.size(), 1u); + const auto live_id = engine.starts[0].id; + + // Same size and etag: not changed. + vdm::net::ProbeResult pr; + pr.effective_url = "https://cdn.example/new-signed-url"; + pr.total_size = 1000; + pr.etag = "\"same\""; + pr.resumable = true; + engine.auto_probe_result = vdm::Result{pr}; + + std::optional> got; + sched.refresh_url("ru0", "https://cdn.example/new-signed-url", std::nullopt, std::nullopt, + [&](auto r2) { got = std::move(r2); }); + CHECK(got.has_value()); + CHECK(got->has_value()); + if (got && *got) { + CHECK((*got)->ok); + CHECK(!(*got)->contentChanged); + CHECK((*got)->resumable); + } + // The live handle got its URL swapped, not a fresh start(). + CHECK_EQ(engine.starts.size(), 1u); + CHECK_EQ(engine.refreshed_urls.size(), std::size_t{1}); + if (!engine.refreshed_urls.empty()) { + CHECK_EQ(engine.refreshed_urls[0].id.value, live_id.value); + CHECK_EQ(engine.refreshed_urls[0].url, std::string("https://cdn.example/new-signed-url")); + } + auto row = tasks.get("ru0").value().value(); + CHECK_EQ(row.url, std::string("https://cdn.example/new-signed-url")); + + // A different size: content changed. + pr.total_size = 2000; + engine.auto_probe_result = vdm::Result{pr}; + std::optional> got2; + sched.refresh_url("ru0", "https://cdn.example/another-url", std::nullopt, std::nullopt, + [&](auto r2) { got2 = std::move(r2); }); + CHECK(got2.has_value() && got2->has_value()); + if (got2 && *got2) CHECK((*got2)->contentChanged); + + // Unknown task: TaskNotFound, no probe issued. + const auto probes_before = engine.probe_requests.size(); + std::optional> got3; + sched.refresh_url("does-not-exist", "https://x/y", std::nullopt, std::nullopt, + [&](auto r2) { got3 = std::move(r2); }); + CHECK(got3.has_value() && !got3->has_value()); + if (got3 && !*got3) CHECK(got3->error().code == velox::proto::ErrorCode::TaskNotFound); + CHECK_EQ(engine.probe_requests.size(), probes_before); + + tasks.remove("ru0"); + } } TEST_MAIN()