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

daemon: capture.offer for real (D7/D8) — rules, category resolution, dedupe, deadline

Browse Source 
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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP
This commit is contained in:
sami
2026-09-12 13:58:45 +04:00
co-authored by Claude Sonnet 5
parent e0edf7a084
commit 6632b75099
14 changed files with 696 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/src/rpc/capture_data_source.hpp
+50
View File
@@ -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 <chrono>
#include <cstdint>
#include <string>
#include <vector>
#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<std::string> monitored_extensions() = 0;
virtual std::vector<std::string> monitored_mime_types() = 0;
virtual std::int64_t min_size_bytes() = 0;
virtual std::vector<std::string> excluded_hosts() = 0;
// Enabled rules, in priority order — ready for vdm::rules::match_rules as-is.
virtual std::vector<vdm::rules::Rule> 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
daemon/src/rpc/dispatcher.cpp
+242 -4
View File
@@ -1,6 +1,7 @@
#include "rpc/dispatcher.hpp"
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <filesystem>
#include <limits>
@@ -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<char>(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<char>(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<char>(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<std::uint64_t>(*r.match.minSizeBytes);
if (r.match.maxSizeBytes) out.match.max_size_bytes = static_cast<std::uint64_t>(*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<std::uint32_t>(*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<std::string> monitored_extensions() override {
return store::Settings(db_).get_string_array("capture.monitoredExtensions");
}
std::vector<std::string> 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<std::string> excluded_hosts() override {
return store::Settings(db_).get_string_array("capture.excludedHosts");
}
std::vector<vdm::rules::Rule> enabled_rules() override {
std::vector<vdm::rules::Rule> 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<proto::CaptureRules>
VeloxDispatcher::on_capture_getRules(const proto::CaptureGetRulesParams&) {
return not_implemented<proto::CaptureRules>("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<proto::CaptureOfferResult>
VeloxDispatcher::on_capture_offer(const proto::CaptureOfferParams&) {
return not_implemented<proto::CaptureOfferResult>("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<proto::CaptureOfferResultReason> reason) {
proto::CaptureOfferResult r;
r.action = proto::CaptureOfferResultAction::Ignore;
r.reason = reason;
return proto::HandlerResult<proto::CaptureOfferResult>(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<std::uint64_t>(*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<std::int64_t>(*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<proto::CategoryListResult>
VeloxDispatcher::on_category_list(const proto::CategoryListParams&) {
daemon/src/rpc/dispatcher.hpp
+8
View File
@@ -14,6 +14,7 @@
#include <functional>
#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<void()> 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<velox::proto::CaptureRules>
on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override;
velox::proto::HandlerResult<velox::proto::CaptureOfferResult>
@@ -126,6 +133,7 @@ private:
EventHub& hub_;
TaskActionPort* actions_;
std::function<void()> on_mutation_;
CaptureDataSource* capture_source_for_test_ = nullptr;
};
} // namespace velox::daemon::rpc