From 4f6fb1029df6b7cf85ef73ea6115c623c94fa1ba Mon Sep 17 00:00:00 2001 From: sami Date: Fri, 11 Sep 2026 17:30:41 +0400 Subject: [PATCH] =?UTF-8?q?daemon:=20D2=20=E2=80=94=20download.probe,=20ge?= =?UTF-8?q?nuinely=20async=20on=20both=20transports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit download.probe was a stub (-32603). It needs an HTTP round trip on the engine's probe pool (up to the schema's 30s x-deadlineMs), which cannot fit VeloxDispatcher's synchronous on_download_probe -> HandlerResult return without blocking the RPC loop for the duration — a hard no per CLAUDE.md ("never block the RPC loop") and AGENT-DAEMON.md build step 1. uds_server.cpp and ws_server.cpp special-case "download.probe" before the generic dispatch(), exactly the way they already special-case session.hello/session.subscribe: parse the params, call the port, and queue the reply whenever the callback fires (dropped silently if the connection is gone by then). rpc::TaskActionPort gains probe_now(DownloadProbeParams, callback) — kept in proto/std terms, no vdm::net::* in the signature, so veloxd_rpc never needs core/include's vdm headers just to declare this. sched::Scheduler::probe_now is the implementation: builds a vdm::net::ProbeRequest, runs it on the engine's probe pool, marshals the engine-thread callback back onto the loop (deps_.post_to_loop, same as every other engine callback here), maps a probe failure to -32013 ProbeFailed (data.httpStatus set when there was an HTTP response), and fills suggestedCategoryId/suggestedSaveDir with a plain extension match against the categories table — not the real rules engine, which is still D3; noted in a comment. Verified against real veloxd + tools/testserver, not just unit tests: a real probe answers in ~5ms with size/resumable/etag/redirect chain; a bad host maps to -32013; and — the actual point of the async design — a connection running a 10s slow-loris probe does not block a second connection's download.list, which answers in ~1ms while the probe is still outstanding. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP --- daemon/docs/deferrals.md | 2 +- daemon/src/main.cpp | 4 +- daemon/src/rpc/task_action_port.hpp | 19 +++++++ daemon/src/rpc/uds_server.cpp | 40 +++++++++++++- daemon/src/rpc/uds_server.hpp | 14 ++++- daemon/src/rpc/ws_server.cpp | 39 ++++++++++++- daemon/src/rpc/ws_server.hpp | 12 +++- daemon/src/sched/scheduler.cpp | 80 ++++++++++++++++++++++++++- daemon/src/sched/scheduler.hpp | 18 +++--- daemon/tests/sched_scheduler_test.cpp | 45 +++++++++++++++ 10 files changed, 255 insertions(+), 18 deletions(-) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 52aad48..3d3d8e7 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -6,7 +6,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | What | Where | Why deferred | Closes when | |---|---|---|---|---| | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) | -| D2 | `download.probe` → `-32603` | `rpc/dispatcher.cpp` | `download.add` is wired (`fs/safepath` + store, real `-32011`); `download.probe` needs the engine's probe path for `-32013` | probe with the engine link (CORE stage 3 is landed; wire `Engine::probe`) | +| ~~D2~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done | | D3 | Stub handlers for the rest: `download.remove/addBatch/refreshUrl/provideAuth/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder`, `category.upsert/remove`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list`, `queue.list`, `queue.start`/`stop` are done | Per method, as each wires to the store/scheduler | | ~~D4a~~ | **Closed** — `sched/engine_port_core.hpp` wraps `vdm::Engine` + `segment_budget()`; `main.cpp` constructs `Engine` + `Scheduler`, calls `reconcile_after_restart` / `reload_config` / `tick` at startup | — | — | done (`lane/core` stage 8 merged) | | ~~D4b~~ | **Closed** — `download.pause`/`resume`/`start`/`cancel` and `queue.start`/`stop` all drive the scheduler now, and apply *immediately* (not deferred to the next tick — pausing/resuming/cancelling a live transfer can't wait up to 1s, and per ADR 0013 §3 the governor never touches a user-owned pause on its own). New `rpc::TaskActionPort` interface (owned by `rpc/`, implemented by `sched::Scheduler`) is the seam dispatcher.hpp depends on instead of `sched/scheduler.hpp` directly — avoids a real `veloxd_rpc` <-> `veloxd_sched` circular library dependency (`veloxd_sched` already links `veloxd_rpc` for `EventHub`). `Scheduler::user_pause/resume/start/cancel` + `pause_queue` engine-call-then-eager-transition, matching `tick()`'s existing `to_pause` pattern. Fixed a real bug hit while building this: `transition()` always overwrote `pause_reason` to NULL when the engine's own delayed pause-ack callback arrived with no explicit reason, clobbering whatever the actual initiator (user or governor) had just written — now it preserves the stored reason when none is supplied. Verified against real `veloxd` + `tools/testserver`: pausing a live single-segment throttled transfer freezes `downloadedBytes`, resume continues it from that point, cancel stops it; `queue.stop(pauseRunning:true)` pauses the queue's running task immediately. NOTE: `download.start`'s contract "a task in 'queued' jumps its queue" (priority bump) is not implemented — admission is still plain FIFO by `created_at`. | `sched/scheduler.{cpp,hpp}`, `rpc/task_action_port.hpp`, `rpc/dispatcher.{hpp,cpp}`, `store/queues.{cpp,hpp}` | — | done, except the queue-jump priority bump noted above | diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index 788ce4a..7b1160d 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -161,7 +161,7 @@ int main() { }); } - velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path()); + velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path(), &scheduler); if (const auto ec = uds.start()) { std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message() << "\n"; @@ -175,7 +175,7 @@ int main() { // TODO(build step 7): replace EnvAutoApprover with a GUI-dialog / desktop-notification // approver. Until then pairing needs VELOX_PAIR_AUTO=1. velox::daemon::rpc::EnvAutoApprover approver; - velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt); + velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt, &scheduler); if (const auto ec = ws.start()) { std::cerr << "veloxd: WebSocket transport unavailable (" << ec.message() << "); the extension fallback will not work this run\n"; diff --git a/daemon/src/rpc/task_action_port.hpp b/daemon/src/rpc/task_action_port.hpp index 6f03218..ce39e67 100644 --- a/daemon/src/rpc/task_action_port.hpp +++ b/daemon/src/rpc/task_action_port.hpp @@ -12,9 +12,12 @@ // rpc/, so adding an rpc-defined base costs nothing new); main.cpp hands the dispatcher a // `TaskActionPort*` pointing at the same Scheduler it constructs. +#include #include #include +#include "velox_proto.hpp" + namespace velox::daemon::rpc { class TaskActionPort { @@ -39,6 +42,22 @@ public: // queue.stop(pauseRunning=true): pause every currently-running task in `queue_id` now. // Returns the wire ids actually paused. virtual std::vector pause_queue(const std::string& queue_id) = 0; + + // download.probe (D2), the File Info dialog's own network round trip — no task row + // involved. Genuinely async (the engine's probe pool; up to the schema's 30s + // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe: + // the RPC server layer (uds_server.cpp / ws_server.cpp) special-cases "download.probe" + // before the generic dispatch(), the same way it already special-cases session.hello, + // calls this, and queues the reply whenever `done` fires — on an engine thread, so the + // implementation must marshal back to the loop before calling it, the same as every + // other EnginePort callback. Kept in std::string/proto terms (not vdm::net::*) so this + // header — included by dispatcher.hpp, part of veloxd_rpc — never needs core/include's + // vdm headers; the vdm::net::ProbeRequest/ProbeResult conversion lives in sched/, which + // already depends on vdm. + virtual void probe_now( + const velox::proto::DownloadProbeParams& params, + std::function)> + done) = 0; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp index 5b3c36c..aaf0ac4 100644 --- a/daemon/src/rpc/uds_server.cpp +++ b/daemon/src/rpc/uds_server.cpp @@ -57,8 +57,9 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json } // namespace UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, EventHub& hub, - std::string socket_path) - : loop_(loop), dispatcher_(dispatcher), hub_(hub), path_(std::move(socket_path)) {} + std::string socket_path, TaskActionPort* actions) + : loop_(loop), dispatcher_(dispatcher), hub_(hub), actions_(actions), + path_(std::move(socket_path)) {} UdsServer::~UdsServer() { for (auto& [fd, c] : conns_) { @@ -202,12 +203,47 @@ void UdsServer::handle_line(Conn& c, const std::string& line) { } } + if (method == "download.probe") { + handle_download_probe(c, req); + return; + } + // Everything else: the generated router. It returns a null json for a notification // that needs no reply. json reply = proto::dispatch(dispatcher_, proto::Transport::Uds, req); if (!reply.is_null()) queue_reply(c, reply); } +void UdsServer::handle_download_probe(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + queue_reply(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.probe")); + return; + } + + const int fd = c.fd; + actions_->probe_now( + *parsed, [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + queue_reply(*it->second, proto::make_result(id, *r)); + } else { + queue_reply(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool UdsServer::handle_session_method(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/uds_server.hpp b/daemon/src/rpc/uds_server.hpp index dd88faf..d76f7a8 100644 --- a/daemon/src/rpc/uds_server.hpp +++ b/daemon/src/rpc/uds_server.hpp @@ -22,6 +22,7 @@ #include "rpc/event_hub.hpp" #include "rpc/ndjson.hpp" +#include "rpc/task_action_port.hpp" #include "velox_proto.hpp" namespace velox::daemon::rpc { @@ -30,8 +31,11 @@ class EventLoop; class UdsServer { public: + // `actions` is optional (nullptr in tests that don't need download.probe) — see + // handle_download_probe's own comment for why this method can't go through the + // generic dispatch() path like everything else. UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, EventHub& hub, - std::string socket_path); + std::string socket_path, TaskActionPort* actions = nullptr); ~UdsServer(); UdsServer(const UdsServer&) = delete; @@ -66,6 +70,13 @@ private: bool handle_session_method(Conn& c, const std::string& method, const nlohmann::json& request, nlohmann::json& reply); + // download.probe is genuinely async (up to the schema's 30s x-deadlineMs, on the + // engine's probe pool) and so cannot fit the synchronous generic dispatch() path — + // special-cased here exactly the way handle_session_method special-cases session.*. + // Queues the reply itself, later, when actions_->probe_now()'s callback fires; does + // nothing if the connection is gone by then (client disconnected mid-probe). + void handle_download_probe(Conn& c, const nlohmann::json& request); + void queue_reply(Conn& c, const nlohmann::json& reply); void flush(Conn& c); void close_conn(int fd); @@ -73,6 +84,7 @@ private: EventLoop& loop_; velox::proto::Dispatcher& dispatcher_; EventHub& hub_; + TaskActionPort* actions_; std::string path_; int listen_fd_ = -1; bool bound_ = false; // path_ is ours to unlink on destruction diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp index a3f2aa3..875bee5 100644 --- a/daemon/src/rpc/ws_server.cpp +++ b/daemon/src/rpc/ws_server.cpp @@ -60,12 +60,14 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json } // namespace WsServer::WsServer(EventLoop& loop, proto::Dispatcher& dispatcher, store::Db& db, - PairingApprover& approver, EventHub& hub, RuntimeDir runtime) + PairingApprover& approver, EventHub& hub, RuntimeDir runtime, + TaskActionPort* actions) : loop_(loop), dispatcher_(dispatcher), db_(db), approver_(approver), hub_(hub), + actions_(actions), runtime_(std::move(runtime)) {} WsServer::~WsServer() { @@ -261,10 +263,45 @@ void WsServer::handle_rpc(Conn& c, const std::string& text) { return; } + if (method == "download.probe") { + handle_download_probe(c, req); + return; + } + json reply = proto::dispatch(dispatcher_, proto::Transport::Ws, req); if (!reply.is_null()) send_text(c, reply); } +void WsServer::handle_download_probe(Conn& c, const json& request) { + const json id = request.contains("id") ? request.at("id") : json(nullptr); + const json params_json = request.contains("params") ? request.at("params") : json::object(); + + auto parsed = proto::parse(params_json, "params"); + if (!parsed) { + send_text(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message, + json{{"path", parsed.error().path}})); + return; + } + if (!actions_) { + send_text(c, rpc_error(id, proto::ErrorCode::InternalError, + "not implemented in this build: download.probe")); + return; + } + + const int fd = c.fd; + actions_->probe_now( + *parsed, [this, fd, id](proto::HandlerResult r) { + auto it = conns_.find(fd); + if (it == conns_.end()) return; // client gone while the probe was outstanding + if (r) { + send_text(*it->second, proto::make_result(id, *r)); + } else { + send_text(*it->second, + rpc_error(id, r.error().code, r.error().message, r.error().data)); + } + }); +} + bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json& request, json& reply) { const json id = request.contains("id") ? request.at("id") : json(nullptr); diff --git a/daemon/src/rpc/ws_server.hpp b/daemon/src/rpc/ws_server.hpp index 2f8e8a9..bc71864 100644 --- a/daemon/src/rpc/ws_server.hpp +++ b/daemon/src/rpc/ws_server.hpp @@ -18,6 +18,7 @@ #include "rpc/event_hub.hpp" #include "rpc/pairing.hpp" #include "rpc/runtime_dir.hpp" +#include "rpc/task_action_port.hpp" #include "rpc/ws_frame.hpp" #include "velox_proto.hpp" @@ -31,8 +32,12 @@ class EventLoop; class WsServer { public: + // `actions` is optional (nullptr in tests that don't need download.probe) — see + // handle_download_probe's own comment for why this method can't go through the + // generic dispatch() path like everything else. WsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, velox::daemon::store::Db& db, - PairingApprover& approver, EventHub& hub, RuntimeDir runtime); + PairingApprover& approver, EventHub& hub, RuntimeDir runtime, + TaskActionPort* actions = nullptr); ~WsServer(); WsServer(const WsServer&) = delete; @@ -73,6 +78,10 @@ private: bool handle_session_ws(Conn& c, const std::string& method, const nlohmann::json& request, nlohmann::json& reply); + // See UdsServer::handle_download_probe — same reasoning, same pattern, duplicated per + // transport because each owns its own Conn/send mechanics. + void handle_download_probe(Conn& c, const nlohmann::json& request); + void send_text(Conn& c, const nlohmann::json& value); void send_frame(Conn& c, WsOpcode op, std::string_view payload); void begin_close(Conn& c, std::uint16_t code, std::string_view reason); @@ -84,6 +93,7 @@ private: velox::daemon::store::Db& db_; PairingApprover& approver_; EventHub& hub_; + TaskActionPort* actions_; RuntimeDir runtime_; PairingRateLimiter rate_limiter_; diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index f460a53..ac296b5 100644 --- a/daemon/src/sched/scheduler.cpp +++ b/daemon/src/sched/scheduler.cpp @@ -6,6 +6,7 @@ #include #include "sched/schedule_window.hpp" +#include "store/categories.hpp" #include "store/segments.hpp" #include "store/settings.hpp" #include "store/tasks.hpp" @@ -570,9 +571,82 @@ std::vector Scheduler::pause_queue(const std::string& queue_id) { return paused; } -void Scheduler::probe_now(const vdm::net::ProbeRequest& req, - std::function)> done) { - engine_.probe(req, std::move(done)); +namespace { + +// Extension match against the categories table (categories.extensions, per category.list), +// the same table download.add would consult once rules.* actually exists (D3). Not the +// real rules engine — no host/mime/size clauses — just enough that the File Info dialog's +// preselect isn't always "general". +std::string guess_category_id(store::Db& db, const std::string& filename) { + const auto dot = filename.find_last_of('.'); + if (dot == std::string::npos || dot + 1 >= filename.size()) return "general"; + std::string ext = filename.substr(dot + 1); + std::transform(ext.begin(), ext.end(), ext.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + + store::Categories categories(db); + auto cats = categories.list(); + if (!cats) return "general"; + for (const auto& c : *cats) { + for (const auto& e : c.extensions) + if (e == ext) return c.categoryId; + } + return "general"; +} + +} // namespace + +void Scheduler::probe_now( + const proto::DownloadProbeParams& params, + std::function)> done) { + vdm::net::ProbeRequest req; + req.url = params.url; + if (params.headers) + for (const auto& [k, v] : *params.headers) req.headers.push_back({k, v}); + if (params.cookies) + for (const auto& c : *params.cookies) req.cookies.push_back({c.name, c.value}); + if (params.referrer) req.referrer = *params.referrer; + if (params.userAgent) req.user_agent = *params.userAgent; + + engine_.probe(req, [this, params, done](vdm::Result pr) { + deps_.post_to_loop([this, params, done, pr]() { + if (!pr) { + nlohmann::json data; + if (pr.error().http_status != 0) data["httpStatus"] = pr.error().http_status; + done(std::unexpected(proto::HandlerError{ + proto::ErrorCode::ProbeFailed, pr.error().context, data})); + return; + } + + const std::string filename = vdm::net::suggest_filename(*pr); + + proto::DownloadProbeResult r; + r.filename = filename.empty() ? "download.bin" : filename; + if (pr->total_size) r.sizeBytes = static_cast(*pr->total_size); + r.mime = pr->mime; + r.resumable = pr->resumable; + r.effectiveUrl = pr->effective_url.empty() ? params.url : pr->effective_url; + r.suggestedCategoryId = guess_category_id(db_, r.filename); + if (!pr->etag.empty()) r.etag = pr->etag; + if (!pr->last_modified.empty()) r.lastModified = pr->last_modified; + r.acceptRanges = pr->accept_ranges; + if (!pr->redirect_chain.empty()) r.redirectChain = pr->redirect_chain; + if (pr->requires_auth) r.requiresAuth = true; + + store::Settings settings(db_); + store::Categories categories(db_); + if (auto cats = categories.list()) { + for (const auto& c : *cats) + if (c.categoryId == r.suggestedCategoryId) { + r.suggestedSaveDir = c.saveDir; + break; + } + } + if (!r.suggestedSaveDir) r.suggestedSaveDir = settings.get_string("saveTo.defaultDir"); + + done(r); + }); + }); } } // namespace velox::daemon::sched diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index d9db9ae..29aba9a 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -132,13 +132,17 @@ public: // instead of User. Returns the wire ids actually paused. std::vector pause_queue(const std::string& queue_id) override; - // download.probe's standalone use (File Info dialog, no task row involved): a thin - // passthrough to the engine's own probe pool, outside the segment budget (ADR 0011 - // §5). Never blocks — `done` arrives on an engine thread like every other EnginePort - // callback; the caller (the RPC server layer, not this synchronous dispatcher — see - // rpc/dispatcher.hpp's top comment) is responsible for marshalling the reply back. - void probe_now(const vdm::net::ProbeRequest& req, - std::function)> done); + // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the + // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the + // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a + // plain extension match against the categories table; the real rules engine is D3). + // `done` is called already marshalled onto the loop thread via post_to_loop, same as + // every other engine callback here — the caller never has to know it started on an + // engine thread. + void probe_now( + const velox::proto::DownloadProbeParams& params, + std::function)> + done) override; // Diagnostics / tests. std::optional wire_id_of(vdm::TaskId id) const; diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp index 11f281c..8f90977 100644 --- a/daemon/tests/sched_scheduler_test.cpp +++ b/daemon/tests/sched_scheduler_test.cpp @@ -361,6 +361,51 @@ void run() { tasks.remove("uc1"); CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value()); } + + // --- probe_now: success (with a category guess) and a mapped failure ------------ + { + FakeEnginePort engine; + Scheduler sched(*db, engine, + Governor(GovernorConfig{.max_concurrent_downloads = 10, + .max_active_segments = 32})); + + vdm::net::ProbeResult pr; + pr.effective_url = "https://cdn.example/movie.mp4"; + pr.filename_from_url = "movie.mp4"; // suggest_filename() needs this set; the real + // Prober fills it from the URL path itself + pr.total_size = 123456; + pr.mime = "video/mp4"; + pr.resumable = true; + pr.accept_ranges = true; + pr.etag = "\"abc\""; + engine.auto_probe_result = vdm::Result{pr}; + + velox::proto::DownloadProbeParams params; + params.url = "https://cdn.example/movie.mp4"; + std::optional> got; + sched.probe_now(params, [&](auto r) { got = std::move(r); }); + CHECK(got.has_value()); + CHECK(got->has_value()); + if (got && *got) { + const auto& r = **got; + CHECK_EQ(r.sizeBytes.value_or(-1), std::int64_t{123456}); + CHECK(r.resumable); + CHECK_EQ(r.mime, std::string("video/mp4")); + // movie.mp4 -> the 'video' built-in category by extension. + CHECK_EQ(r.suggestedCategoryId, std::string("video")); + } + + vdm::ErrorInfo err; + err.code = vdm::Error::connect_failed; + err.context = "connection refused"; + engine.auto_probe_result = vdm::Result{err}; + std::optional> got2; + sched.probe_now(params, [&](auto r) { got2 = std::move(r); }); + CHECK(got2.has_value()); + CHECK(!got2->has_value()); + if (got2 && !*got2) + CHECK(got2->error().code == velox::proto::ErrorCode::ProbeFailed); + } } TEST_MAIN()