From 6632b750992ad96d8bb88b039639fc64a2298aa1 Mon Sep 17 00:00:00 2001 From: sami Date: Sat, 12 Sep 2026 13:58:45 +0400 Subject: [PATCH] =?UTF-8?q?daemon:=20capture.offer=20for=20real=20(D7/D8)?= =?UTF-8?q?=20=E2=80=94=20rules,=20category=20resolution,=20dedupe,=20dead?= =?UTF-8?q?line?= 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()