// Scheduler against a FakeEnginePort and an in-memory store: admission, priority order, // queue stop, restart reconciliation, and the engine-state -> store projection. #include #include #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 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(""); } } // 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 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("ev0")); CHECK_EQ(received[0]["params"]["state"].get(), std::string("probing")); CHECK_EQ(received[0]["params"]["previousState"].get(), std::string("queued")); CHECK(received[0]["params"]["error"].is_null()); CHECK_EQ(received[0]["params"]["summary"]["taskId"].get(), 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("probing")); CHECK_EQ(received[1]["params"]["state"].get(), 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("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{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); } // --- refresh_url: content unchanged, content changed, and a live-handle swap ----- { FakeEnginePort engine; Scheduler sched(*db, engine, Governor(GovernorConfig{.max_concurrent_downloads = 10, .max_active_segments = 32})); store::TaskRow r; r.task_id = "ru0"; r.url = "https://cdn.example/old-signed-url"; r.save_dir = "/tmp"; r.filename = "ru0.bin"; r.state = "queued"; r.created_at = "2026-09-12T00:00:00Z"; r.size_bytes = 1000; r.etag = "\"same\""; CHECK(tasks.insert(r).has_value()); CHECK(sched.tick().has_value()); CHECK_EQ(engine.starts.size(), 1u); const auto live_id = engine.starts[0].id; // Same size and etag: not changed. vdm::net::ProbeResult pr; pr.effective_url = "https://cdn.example/new-signed-url"; pr.total_size = 1000; pr.etag = "\"same\""; pr.resumable = true; engine.auto_probe_result = vdm::Result{pr}; std::optional> got; sched.refresh_url("ru0", "https://cdn.example/new-signed-url", std::nullopt, std::nullopt, [&](auto r2) { got = std::move(r2); }); CHECK(got.has_value()); CHECK(got->has_value()); if (got && *got) { CHECK((*got)->ok); CHECK(!(*got)->contentChanged); CHECK((*got)->resumable); } // The live handle got its URL swapped, not a fresh start(). CHECK_EQ(engine.starts.size(), 1u); CHECK_EQ(engine.refreshed_urls.size(), std::size_t{1}); if (!engine.refreshed_urls.empty()) { CHECK_EQ(engine.refreshed_urls[0].id.value, live_id.value); CHECK_EQ(engine.refreshed_urls[0].url, std::string("https://cdn.example/new-signed-url")); } auto row = tasks.get("ru0").value().value(); CHECK_EQ(row.url, std::string("https://cdn.example/new-signed-url")); // A different size: content changed. pr.total_size = 2000; engine.auto_probe_result = vdm::Result{pr}; std::optional> got2; sched.refresh_url("ru0", "https://cdn.example/another-url", std::nullopt, std::nullopt, [&](auto r2) { got2 = std::move(r2); }); CHECK(got2.has_value() && got2->has_value()); if (got2 && *got2) CHECK((*got2)->contentChanged); // Unknown task: TaskNotFound, no probe issued. const auto probes_before = engine.probe_requests.size(); std::optional> got3; sched.refresh_url("does-not-exist", "https://x/y", std::nullopt, std::nullopt, [&](auto r2) { got3 = std::move(r2); }); CHECK(got3.has_value() && !got3->has_value()); if (got3 && !*got3) CHECK(got3->error().code == velox::proto::ErrorCode::TaskNotFound); CHECK_EQ(engine.probe_requests.size(), probes_before); tasks.remove("ru0"); } } TEST_MAIN()