samiandClaude Sonnet 5
4f6fb1029d
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<T> 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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP
412 lines
18 KiB
C++
412 lines
18 KiB
C++
// Scheduler against a FakeEnginePort and an in-memory store: admission, priority order,
|
|
// queue stop, restart reconciliation, and the engine-state -> store projection.
|
|
|
|
#include <string>
|
|
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include "check.hpp"
|
|
#include "rpc/event_hub.hpp"
|
|
#include "sched/fake_engine_port.hpp"
|
|
#include "sched/governor.hpp"
|
|
#include "sched/scheduler.hpp"
|
|
#include "store/migrations.hpp"
|
|
#include "store/settings.hpp"
|
|
#include "store/sqlite.hpp"
|
|
#include "store/tasks.hpp"
|
|
|
|
using namespace velox::daemon;
|
|
using sched::FakeEnginePort;
|
|
using sched::Governor;
|
|
using sched::GovernorConfig;
|
|
using sched::Scheduler;
|
|
|
|
namespace {
|
|
|
|
store::TaskRow task(std::string id, std::string state, std::string created,
|
|
std::optional<std::string> 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 = std::move(state);
|
|
r.created_at = std::move(created);
|
|
r.queue_id = std::move(queue);
|
|
if (r.queue_id) r.queue_position = pos;
|
|
return r;
|
|
}
|
|
|
|
std::string task_state(store::Db& db, const std::string& id) {
|
|
store::Tasks t(db);
|
|
auto g = t.get(id);
|
|
return (g && *g) ? (*g)->state : std::string("<none>");
|
|
}
|
|
|
|
} // namespace
|
|
|
|
void run() {
|
|
auto db = store::Db::open(":memory:");
|
|
CHECK(db.has_value());
|
|
if (!db) return;
|
|
CHECK(store::migrate_to_head(*db).has_value());
|
|
store::Tasks tasks(*db);
|
|
|
|
// --- admission: 4 queued, cap 2 -> start the 2 oldest, in order -----------------
|
|
{
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 2,
|
|
.max_active_segments = 32}));
|
|
for (int i = 0; i < 4; ++i)
|
|
CHECK(tasks.insert(task("a" + std::to_string(i), "queued",
|
|
"2026-09-10T10:0" + std::to_string(i) + ":00Z"))
|
|
.has_value());
|
|
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 2u);
|
|
CHECK_EQ(engine.starts[0].url, std::string("https://cdn.example/a0"));
|
|
CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/a1"));
|
|
CHECK_EQ(engine.starts[0].save_path, std::string("/tmp/a0.bin"));
|
|
CHECK_EQ(task_state(*db, "a0"), std::string("probing"));
|
|
CHECK_EQ(task_state(*db, "a2"), std::string("queued")); // not admitted
|
|
|
|
// set_task_order carries exactly the started tasks, oldest first.
|
|
CHECK_EQ(engine.last_order().size(), 2u);
|
|
CHECK(engine.last_order()[0] == engine.starts[0].id);
|
|
|
|
// A second tick with no free slots starts nothing new.
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 2u);
|
|
}
|
|
|
|
// --- engine reports downloading, then one completes -> a slot frees ------------
|
|
{
|
|
for (const char* id : {"a0", "a1", "a2", "a3"}) tasks.remove(id);
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 1,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("b0", "queued", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(tasks.insert(task("b1", "queued", "2026-09-10T10:01:00Z")).has_value());
|
|
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 1u); // b0 only
|
|
CHECK_EQ(task_state(*db, "b0"), std::string("probing"));
|
|
|
|
sched.on_engine_state("b0", "probing", "downloading", std::nullopt);
|
|
CHECK_EQ(task_state(*db, "b0"), std::string("downloading"));
|
|
|
|
sched.on_engine_state("b0", "downloading", "complete", std::nullopt);
|
|
CHECK_EQ(task_state(*db, "b0"), std::string("complete"));
|
|
|
|
CHECK(sched.tick().has_value()); // b0 terminal -> b1 admitted
|
|
CHECK_EQ(engine.starts.size(), 2u);
|
|
CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/b1"));
|
|
}
|
|
|
|
// --- a stopped queue: running tasks get paused with reason queue_stopped -------
|
|
{
|
|
for (const char* id : {"b0", "b1"}) tasks.remove(id);
|
|
CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("q0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value());
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 1u);
|
|
sched.on_engine_state("q0", "connecting", "downloading", std::nullopt);
|
|
|
|
CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value());
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.paused.size(), 1u);
|
|
CHECK_EQ(task_state(*db, "q0"), std::string("paused"));
|
|
store::Tasks t(*db);
|
|
CHECK_EQ(t.get("q0").value().value().pause_reason.value_or(""), std::string("queue_stopped"));
|
|
|
|
// Restart the queue -> the task resumes (not a fresh start).
|
|
CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.resumed.size(), 1u);
|
|
CHECK_EQ(engine.starts.size(), 1u); // no new start
|
|
}
|
|
|
|
// --- an engine auto-pause (error present) -> pause_reason 'auto', not touched --
|
|
{
|
|
for (const char* id : {"q0"}) tasks.remove(id);
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("auth", "queued", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(sched.tick().has_value());
|
|
|
|
sched::TaskErrorFields ef;
|
|
ef.code = "auth_required";
|
|
ef.message = "401";
|
|
ef.http_status = 401;
|
|
sched.on_engine_state("auth", "connecting", "paused", ef);
|
|
|
|
store::Tasks t(*db);
|
|
auto row = t.get("auth").value().value();
|
|
CHECK_EQ(row.state, std::string("paused"));
|
|
CHECK_EQ(row.pause_reason.value_or(""), std::string("auto"));
|
|
CHECK_EQ(row.error_code.value_or(""), std::string("auth_required"));
|
|
|
|
// A tick must NOT resume an auto-paused task.
|
|
engine.resumed.clear();
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.resumed.size(), 0u);
|
|
}
|
|
|
|
// --- reconcile_after_restart: CORE-owned states -> queued --------------------
|
|
{
|
|
for (const char* id : {"auth"}) tasks.remove(id);
|
|
CHECK(tasks.insert(task("r0", "downloading", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(tasks.insert(task("r1", "verifying", "2026-09-10T10:01:00Z")).has_value());
|
|
CHECK(tasks.insert(task("r2", "paused", "2026-09-10T10:02:00Z")).has_value());
|
|
CHECK(db->exec("UPDATE tasks SET pause_reason='user' WHERE task_id='r2'").has_value());
|
|
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine, Governor(GovernorConfig{}));
|
|
CHECK(sched.reconcile_after_restart().has_value());
|
|
CHECK_EQ(task_state(*db, "r0"), std::string("queued"));
|
|
CHECK_EQ(task_state(*db, "r1"), std::string("queued"));
|
|
CHECK_EQ(task_state(*db, "r2"), std::string("paused")); // paused survives
|
|
store::Tasks t(*db);
|
|
CHECK_EQ(t.get("r2").value().value().pause_reason.value_or(""), std::string("user"));
|
|
}
|
|
|
|
// --- reload_config pushes the caps to the engine ---------------------------
|
|
{
|
|
store::Settings settings(*db);
|
|
CHECK(settings.set_raw("connection.maxActiveSegments", "12").has_value());
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine, Governor(GovernorConfig{}));
|
|
CHECK(sched.reload_config().has_value());
|
|
CHECK_EQ(engine.max_active_segments.size(), 1u);
|
|
CHECK_EQ(engine.max_active_segments.back(), 12u);
|
|
CHECK_EQ(sched.reload_config().has_value() ? 0 : 1, 0);
|
|
}
|
|
|
|
// --- event.task.state: published on admission, on a paused transition, and on an
|
|
// engine-reported transition; previousState reflects the store's prior row ----------
|
|
{
|
|
for (const char* id : {"r0", "r1", "r2"}) tasks.remove(id); // leftover from above
|
|
FakeEnginePort engine;
|
|
rpc::EventHub hub;
|
|
Scheduler sched(*db, engine, Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}),
|
|
&hub);
|
|
std::vector<nlohmann::json> received;
|
|
const auto sub = hub.subscribe([&](const nlohmann::json& n) { received.push_back(n); });
|
|
hub.set_filter(sub, {velox::proto::Event::TaskState}, std::nullopt);
|
|
|
|
CHECK(tasks.insert(task("ev0", "queued", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(sched.tick().has_value()); // admits ev0: queued -> probing
|
|
CHECK_EQ(received.size(), 1u);
|
|
CHECK_EQ(received[0]["params"]["taskId"].get<std::string>(), std::string("ev0"));
|
|
CHECK_EQ(received[0]["params"]["state"].get<std::string>(), std::string("probing"));
|
|
CHECK_EQ(received[0]["params"]["previousState"].get<std::string>(), std::string("queued"));
|
|
CHECK(received[0]["params"]["error"].is_null());
|
|
CHECK_EQ(received[0]["params"]["summary"]["taskId"].get<std::string>(), std::string("ev0"));
|
|
|
|
sched.on_engine_state("ev0", "probing", "downloading", std::nullopt);
|
|
CHECK_EQ(received.size(), 2u);
|
|
CHECK_EQ(received[1]["params"]["previousState"].get<std::string>(), std::string("probing"));
|
|
CHECK_EQ(received[1]["params"]["state"].get<std::string>(), std::string("downloading"));
|
|
|
|
sched::TaskErrorFields ef;
|
|
ef.code = "connection_reset";
|
|
ef.message = "reset";
|
|
sched.on_engine_state("ev0", "downloading", "paused", ef);
|
|
CHECK_EQ(received.size(), 3u);
|
|
CHECK(!received[2]["params"]["error"].is_null());
|
|
CHECK_EQ(received[2]["params"]["error"]["code"].get<std::string>(),
|
|
std::string("connection_reset"));
|
|
|
|
tasks.remove("ev0");
|
|
}
|
|
|
|
// --- progress_snapshot: only started tasks, plus a store side-effect --------------
|
|
{
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("pr0", "queued", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 1u);
|
|
|
|
vdm::task::Progress p;
|
|
p.downloaded = 12345;
|
|
p.speed_bps = 999;
|
|
p.effective_segments = 4;
|
|
p.effective_buffer_bytes = 65536;
|
|
engine.fake_progress[engine.starts[0].id.value] = p;
|
|
|
|
const auto snap = sched.progress_snapshot();
|
|
CHECK_EQ(snap.size(), 1u);
|
|
CHECK_EQ(snap[0].task_id, std::string("pr0"));
|
|
CHECK_EQ(snap[0].downloaded_bytes, 12345u);
|
|
CHECK_EQ(snap[0].speed_bps, 999u);
|
|
|
|
auto row = tasks.get("pr0").value().value();
|
|
CHECK_EQ(row.downloaded_bytes, 12345);
|
|
CHECK_EQ(row.eff_segments, 4);
|
|
CHECK((row.eff_buffer_bytes.has_value() && *row.eff_buffer_bytes == 65536));
|
|
|
|
tasks.remove("pr0");
|
|
}
|
|
|
|
// --- user_pause / user_resume: a live task, and one that never started -----------
|
|
{
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("up0", "queued", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(tasks.insert(task("up1", "paused", "2026-09-10T10:00:00Z")).has_value());
|
|
CHECK(sched.tick().has_value()); // admits up0; up1 stays paused (governor never
|
|
// touches a user-owned pause)
|
|
CHECK_EQ(engine.starts.size(), 1u);
|
|
const auto live_id = engine.starts[0].id;
|
|
|
|
// Pausing a live task calls the engine now and transitions eagerly — not left for
|
|
// the next tick.
|
|
auto r = sched.user_pause("up0");
|
|
CHECK(r.found);
|
|
CHECK(r.changed);
|
|
CHECK_EQ(r.state, std::string("paused"));
|
|
CHECK_EQ(engine.paused.size(), 1u);
|
|
CHECK_EQ(engine.paused[0].value, live_id.value);
|
|
CHECK_EQ(task_state(*db, "up0"), std::string("paused"));
|
|
auto row = tasks.get("up0").value().value();
|
|
CHECK_EQ(row.pause_reason.value_or(""), std::string("user"));
|
|
|
|
// Idempotent: pausing an already-paused task is a no-op, not an error.
|
|
auto again = sched.user_pause("up0");
|
|
CHECK(again.found);
|
|
CHECK(!again.changed);
|
|
|
|
// Resuming a task that still holds a live engine handle calls engine.resume() and
|
|
// goes straight to `connecting`.
|
|
auto res = sched.user_resume("up0");
|
|
CHECK(res.found);
|
|
CHECK(res.changed);
|
|
CHECK_EQ(res.state, std::string("connecting"));
|
|
CHECK_EQ(engine.resumed.size(), 1u);
|
|
CHECK_EQ(engine.resumed[0].value, live_id.value);
|
|
|
|
// The engine's own delayed pause-ack (on_state with no error, arriving after the
|
|
// eager transition already wrote the real reason) must not clobber pause_reason
|
|
// back to NULL.
|
|
(void)sched.user_pause("up0");
|
|
sched.on_engine_state("up0", "downloading", "paused", std::nullopt);
|
|
auto row2 = tasks.get("up0").value().value();
|
|
CHECK_EQ(row2.pause_reason.value_or(""), std::string("user"));
|
|
|
|
// up1 never started (still parked, no engine handle): resume just re-queues it for
|
|
// the next tick's normal admission.
|
|
auto res2 = sched.user_resume("up1");
|
|
CHECK(res2.found);
|
|
CHECK(res2.changed);
|
|
CHECK_EQ(res2.state, std::string("queued"));
|
|
CHECK(engine.resumed.size() == 1u); // up1 was never mapped; no engine call
|
|
|
|
// Not found: a bogus id reports found=false, not a crash.
|
|
auto missing = sched.user_pause("does-not-exist");
|
|
CHECK(!missing.found);
|
|
|
|
tasks.remove("up0");
|
|
tasks.remove("up1");
|
|
}
|
|
|
|
// --- user_cancel + pause_queue --------------------------------------------------
|
|
{
|
|
CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
CHECK(tasks.insert(task("uc0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value());
|
|
CHECK(tasks.insert(task("uc1", "queued", "2026-09-10T10:00:01Z", "main", 1)).has_value());
|
|
CHECK(sched.tick().has_value());
|
|
CHECK_EQ(engine.starts.size(), 2u);
|
|
|
|
auto c = sched.user_cancel("uc0", /*discard_partial=*/true);
|
|
CHECK(c.found);
|
|
CHECK(c.changed);
|
|
CHECK_EQ(c.state, std::string("cancelled"));
|
|
CHECK_EQ(engine.cancelled.size(), 1u);
|
|
CHECK(engine.cancelled[0].second); // discard_partial passed through
|
|
CHECK_EQ(task_state(*db, "uc0"), std::string("cancelled"));
|
|
|
|
// Cancelling an already-terminal task is a no-op.
|
|
auto c2 = sched.user_cancel("uc0", false);
|
|
CHECK(c2.found);
|
|
CHECK(!c2.changed);
|
|
|
|
// pause_queue pauses every still-running task in the queue (uc0 is terminal, so
|
|
// only uc1 is affected) and reports pause_reason 'queue_stopped'.
|
|
const auto paused_ids = sched.pause_queue("main");
|
|
CHECK_EQ(paused_ids.size(), std::size_t{1});
|
|
CHECK_EQ(paused_ids[0], std::string("uc1"));
|
|
CHECK_EQ(task_state(*db, "uc1"), std::string("paused"));
|
|
auto row = tasks.get("uc1").value().value();
|
|
CHECK_EQ(row.pause_reason.value_or(""), std::string("queue_stopped"));
|
|
|
|
tasks.remove("uc0");
|
|
tasks.remove("uc1");
|
|
CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value());
|
|
}
|
|
|
|
// --- probe_now: success (with a category guess) and a mapped failure ------------
|
|
{
|
|
FakeEnginePort engine;
|
|
Scheduler sched(*db, engine,
|
|
Governor(GovernorConfig{.max_concurrent_downloads = 10,
|
|
.max_active_segments = 32}));
|
|
|
|
vdm::net::ProbeResult pr;
|
|
pr.effective_url = "https://cdn.example/movie.mp4";
|
|
pr.filename_from_url = "movie.mp4"; // suggest_filename() needs this set; the real
|
|
// Prober fills it from the URL path itself
|
|
pr.total_size = 123456;
|
|
pr.mime = "video/mp4";
|
|
pr.resumable = true;
|
|
pr.accept_ranges = true;
|
|
pr.etag = "\"abc\"";
|
|
engine.auto_probe_result = vdm::Result<vdm::net::ProbeResult>{pr};
|
|
|
|
velox::proto::DownloadProbeParams params;
|
|
params.url = "https://cdn.example/movie.mp4";
|
|
std::optional<velox::proto::HandlerResult<velox::proto::DownloadProbeResult>> got;
|
|
sched.probe_now(params, [&](auto r) { got = std::move(r); });
|
|
CHECK(got.has_value());
|
|
CHECK(got->has_value());
|
|
if (got && *got) {
|
|
const auto& r = **got;
|
|
CHECK_EQ(r.sizeBytes.value_or(-1), std::int64_t{123456});
|
|
CHECK(r.resumable);
|
|
CHECK_EQ(r.mime, std::string("video/mp4"));
|
|
// movie.mp4 -> the 'video' built-in category by extension.
|
|
CHECK_EQ(r.suggestedCategoryId, std::string("video"));
|
|
}
|
|
|
|
vdm::ErrorInfo err;
|
|
err.code = vdm::Error::connect_failed;
|
|
err.context = "connection refused";
|
|
engine.auto_probe_result = vdm::Result<vdm::net::ProbeResult>{err};
|
|
std::optional<velox::proto::HandlerResult<velox::proto::DownloadProbeResult>> got2;
|
|
sched.probe_now(params, [&](auto r) { got2 = std::move(r); });
|
|
CHECK(got2.has_value());
|
|
CHECK(!got2->has_value());
|
|
if (got2 && !*got2)
|
|
CHECK(got2->error().code == velox::proto::ErrorCode::ProbeFailed);
|
|
}
|
|
}
|
|
|
|
TEST_MAIN()
|