samiandClaude Sonnet 5
4f6c0cc9d2
rules.list/rules.upsert: new store/rules.{hpp,cpp} (list in priority order; apply()
upserts+removes atomically, generating an id when absent, per the schema's own "never
leaves the table in a half-valid state"). Found and fixed along the way: migration
0001's rules table had no column for Rule.name at all — every rules.list/.upsert call
failed outright ("no such column: name"), unit tests included, since :memory: migrates
through the same path. Migration 0004 adds it.
queue.reorder: new store::Queues::reorder — taskIds must be an exact permutation of
the queue's current membership (compared as sorted sets) or nothing is written and
-32602 names the queue; a valid permutation rewrites every member's queue_position in
one transaction.
schedule.get/schedule.set: a thin wrapper over the queues.schedule column (already
read since D3b, never independently settable). nextRunAt is deliberately left unset —
computing it needs the same local-time, DST-aware window logic
sched/schedule_window.hpp's window_open() only has half of; called out rather than
approximated, and the field is optional.
limiter.get/limiter.set: backed by the same downloads.speedLimitEnabled/
downloads.speedLimitBps settings keys D9 already wired — one bag of truth, not two.
The new part is reaching the engine: EnginePort/TaskActionPort gain
set_global_speed_limit(bps) (0 = unlimited, TokenBucket's own convention), wired to
Engine::rate_limiter().set_global_limit(). Pushed live on every limiter.set *and* on
Scheduler::reload_config() so a limit from a previous run isn't silently unlimited
again after a restart. applyToRunning is accepted but has no lever to pull
differently — a single shared global bucket has no "next task only" variant. Also
found, not chased further: the schema's "globalBps:0 with enabled:true means 'stop
everything'" is the opposite of what TokenBucket does with rate_bps==0 (unlimited) —
a real discrepancy, but the schema says the GUI must not offer that combination.
download.update: "moving saveDir or filename moves the file on disk in the same
operation" — resolved and root-checked like download.add's destination, then the
.veloxpart/.veloxpart.meta pair (or the finished file, if complete) is moved via
rename, falling back to copy+remove across filesystems, only when the resolved
location actually differs. categoryId/queueId(appended to the new queue's run
order)/description/segments/bufferBytes/checksum apply through new
store::Tasks::apply_update.
download.refreshUrl: same async server-layer special-case as download.probe (a real
network round trip, same 30s deadline). Re-probes, flags contentChanged only when
size or validator are both known and actually differ, persists the new URL and probe
result, and swaps the URL on a live engine handle via a newly-widened
EnginePort::refresh_url (now takes headers too, matching DownloadHandle's real
signature — the seam had silently dropped them).
Found and documented, not fixed: the generated parser collapses "field absent" and
"field explicitly null" to the same nullopt for every optional<T> patch field
(DownloadUpdateParamsPatch, Settings) — both schemas document "an explicit null
clears the field" but neither handler can act on it because the wire distinction is
already gone by the time either sees the parsed struct. A generator-level gap
(PROTO's), not something to hand-route around locally.
Verified against real veloxd + tools/testserver: rules create/list, limiter.set
takes effect and reads back, schedule.set/get round-trips, queue.reorder against real
membership (and rejects a non-permutation), download.update renames+recategorizes a
task, download.refreshUrl swaps a paused task's URL and reports contentChanged
correctly. Full ctest: 55/55 (excluding the pre-existing, unrelated conformance
failure noted two commits back).
Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP
721 lines
31 KiB
C++
721 lines
31 KiB
C++
#include "sched/scheduler.hpp"
|
|
|
|
#include <algorithm>
|
|
#include <cctype>
|
|
|
|
#include <nlohmann/json.hpp>
|
|
|
|
#include "sched/schedule_window.hpp"
|
|
#include "store/categories.hpp"
|
|
#include "store/segments.hpp"
|
|
#include "store/settings.hpp"
|
|
#include "store/tasks.hpp"
|
|
|
|
namespace velox::daemon::sched {
|
|
|
|
namespace proto = velox::proto;
|
|
|
|
namespace {
|
|
|
|
std::tm local_now_default() {
|
|
const std::time_t t = std::time(nullptr);
|
|
std::tm tm{};
|
|
::localtime_r(&t, &tm);
|
|
return tm;
|
|
}
|
|
|
|
// host[:port] out of a URL, lowercased. "" when it cannot be parsed (opts the task out of
|
|
// the per-host cap rather than lumping unrelated tasks under "").
|
|
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;
|
|
}
|
|
|
|
// ISO-8601 timestamp -> a monotonic integer for FIFO tiebreaking ("2026-09-10T15:57:50Z"
|
|
// -> 20260910155750). Lexical order of the digits is chronological.
|
|
std::int64_t rank_of(std::string_view iso) {
|
|
std::string digits;
|
|
for (char c : iso)
|
|
if (std::isdigit(static_cast<unsigned char>(c))) digits.push_back(c);
|
|
digits.resize(std::min<std::size_t>(digits.size(), 17)); // fits in int64
|
|
return digits.empty() ? 0 : std::stoll(digits);
|
|
}
|
|
|
|
RunState run_state_of(const std::string& s) {
|
|
if (s == "queued") return RunState::Queued;
|
|
if (s == "paused") return RunState::Paused;
|
|
if (s == "complete" || s == "failed" || s == "cancelled") return RunState::Terminal;
|
|
return RunState::Running; // probing / connecting / downloading / retry_wait / assembling / verifying
|
|
}
|
|
|
|
std::optional<PauseReason> pause_reason_of(const std::optional<std::string>& s) {
|
|
if (!s) return std::nullopt;
|
|
if (*s == "user") return PauseReason::User;
|
|
if (*s == "schedule") return PauseReason::Schedule;
|
|
if (*s == "queue_stopped") return PauseReason::QueueStopped;
|
|
if (*s == "admission_reconcile") return PauseReason::AdmissionReconcile;
|
|
if (*s == "auto") return PauseReason::Auto;
|
|
return std::nullopt;
|
|
}
|
|
|
|
const char* pause_reason_str(PauseReason r) {
|
|
switch (r) {
|
|
case PauseReason::User: return "user";
|
|
case PauseReason::Schedule: return "schedule";
|
|
case PauseReason::QueueStopped: return "queue_stopped";
|
|
case PauseReason::AdmissionReconcile: return "admission_reconcile";
|
|
case PauseReason::Auto: return "auto";
|
|
}
|
|
return "user";
|
|
}
|
|
|
|
const char* engine_state_name(vdm::task::EngineState s) {
|
|
using E = vdm::task::EngineState;
|
|
switch (s) {
|
|
case E::probing: return "probing";
|
|
case E::connecting: return "connecting";
|
|
case E::downloading: return "downloading";
|
|
case E::paused: return "paused";
|
|
case E::retry_wait: return "retry_wait";
|
|
case E::assembling: return "assembling";
|
|
case E::verifying: return "verifying";
|
|
case E::complete: return "complete";
|
|
case E::failed: return "failed";
|
|
case E::cancelled: return "cancelled";
|
|
}
|
|
return "connecting";
|
|
}
|
|
|
|
// The non-terminal states the scheduler cares about — feeds the tasks.list filter.
|
|
std::vector<proto::TaskState> non_terminal_states() {
|
|
return {proto::TaskState::New, proto::TaskState::Probing,
|
|
proto::TaskState::Queued, proto::TaskState::Connecting,
|
|
proto::TaskState::Downloading, proto::TaskState::Paused,
|
|
proto::TaskState::RetryWait, proto::TaskState::Assembling,
|
|
proto::TaskState::Verifying};
|
|
}
|
|
|
|
TaskErrorFields to_error_fields(const vdm::ErrorInfo& err) {
|
|
TaskErrorFields ef;
|
|
ef.code = std::string(vdm::error_name(err.code)); // matches TaskErrorCode by name (ADR 0010)
|
|
ef.message = err.context;
|
|
if (err.http_status != 0) ef.http_status = err.http_status;
|
|
ef.retryable = err.retryable;
|
|
return ef;
|
|
}
|
|
|
|
std::string segment_state_name(vdm::segment::SegState s) {
|
|
using S = vdm::segment::SegState;
|
|
switch (s) {
|
|
case S::idle: return "pending";
|
|
case S::connecting: return "connecting";
|
|
case S::downloading: return "downloading";
|
|
case S::stalled: return "stalled";
|
|
case S::complete: return "complete";
|
|
case S::failed: return "failed";
|
|
}
|
|
return "pending";
|
|
}
|
|
|
|
} // namespace
|
|
|
|
Scheduler::Scheduler(store::Db& db, EnginePort& engine, Governor governor, rpc::EventHub* hub,
|
|
Deps deps)
|
|
: db_(db), engine_(engine), governor_(std::move(governor)), hub_(hub), deps_(std::move(deps)) {
|
|
if (!deps_.local_now) deps_.local_now = local_now_default;
|
|
if (!deps_.post_to_loop) deps_.post_to_loop = [](std::function<void()> f) { f(); };
|
|
}
|
|
|
|
void Scheduler::map(const std::string& wire_id, vdm::TaskId engine_id) {
|
|
to_engine_[wire_id] = engine_id;
|
|
to_wire_[engine_id] = wire_id;
|
|
}
|
|
void Scheduler::unmap_engine(vdm::TaskId engine_id) {
|
|
if (auto it = to_wire_.find(engine_id); it != to_wire_.end()) {
|
|
to_engine_.erase(it->second);
|
|
to_wire_.erase(it);
|
|
}
|
|
}
|
|
std::optional<std::string> Scheduler::wire_id_of(vdm::TaskId id) const {
|
|
auto it = to_wire_.find(id);
|
|
return it == to_wire_.end() ? std::nullopt : std::optional<std::string>(it->second);
|
|
}
|
|
std::optional<vdm::TaskId> Scheduler::engine_id_of(const std::string& wire_id) const {
|
|
auto it = to_engine_.find(wire_id);
|
|
return it == to_engine_.end() ? std::nullopt : std::optional<vdm::TaskId>(it->second);
|
|
}
|
|
|
|
store::DbResult<void> Scheduler::reconcile_after_restart() {
|
|
// Any CORE-owned state (probing..verifying) becomes `queued`; the engine knows nothing
|
|
// across a restart and will re-probe / re-resume from the .veloxpart.meta sidecar when
|
|
// the scheduler starts it again (ADR 0013 §5). `paused` and `new` are left alone.
|
|
return db_.exec(
|
|
"UPDATE tasks SET state = 'queued', pause_reason = NULL "
|
|
"WHERE state IN ('probing','connecting','downloading','retry_wait','assembling','verifying')");
|
|
}
|
|
|
|
store::DbResult<void> Scheduler::reload_config() {
|
|
store::Settings settings(db_);
|
|
GovernorConfig cfg;
|
|
cfg.max_concurrent_downloads = settings.get_int("connection.maxConcurrentDownloads");
|
|
cfg.max_active_segments = settings.get_int("connection.maxActiveSegments");
|
|
|
|
// Per-host caps: a JSON object {host: n} under a daemon-local key. Absent => none.
|
|
if (auto raw = settings.get_raw("saveTo.hostSegmentCaps"); raw && *raw) {
|
|
auto j = nlohmann::json::parse(**raw, nullptr, false);
|
|
if (j.is_object()) {
|
|
for (const auto& [host, n] : j.items()) {
|
|
if (n.is_number_integer()) {
|
|
const auto cap = n.get<std::int64_t>();
|
|
cfg.host_caps[host] = cap;
|
|
engine_.set_host_segment_cap(host, static_cast<std::uint32_t>(std::max<std::int64_t>(cap, 0)));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
governor_.set_config(cfg);
|
|
engine_.set_max_active_segments(
|
|
static_cast<std::uint32_t>(std::max<std::int64_t>(cfg.max_active_segments, 1)));
|
|
|
|
// The global speed limit persists across a restart the same as any other setting, but
|
|
// (unlike connection.* above) nothing re-derives it into engine state on its own —
|
|
// limiter.set is the only other place that calls set_global_speed_limit, and that only
|
|
// fires on an explicit RPC in a running daemon. Push it here too so a limit set in a
|
|
// previous run is not silently unlimited again after a restart.
|
|
const bool limit_enabled = settings.get_bool("downloads.speedLimitEnabled");
|
|
const std::int64_t limit_bps = settings.get_int("downloads.speedLimitBps");
|
|
engine_.set_global_speed_limit(limit_enabled ? static_cast<std::uint64_t>(std::max<std::int64_t>(limit_bps, 0))
|
|
: 0);
|
|
return {};
|
|
}
|
|
|
|
store::DbResult<void> Scheduler::tick() {
|
|
// --- snapshot: queues -----------------------------------------------------------
|
|
std::vector<QueueView> queues;
|
|
{
|
|
auto st = db_.prepare("SELECT queue_id, state, max_concurrent, schedule FROM queues");
|
|
if (!st) return std::unexpected(st.error());
|
|
const std::tm now = deps_.local_now();
|
|
for (;;) {
|
|
auto row = st->step();
|
|
if (!row) return std::unexpected(row.error());
|
|
if (!*row) break;
|
|
QueueView q;
|
|
q.queue_id = st->column_text(0);
|
|
q.running = st->column_text(1) == "running";
|
|
q.max_concurrent = st->column_int(2);
|
|
if (!st->column_is_null(3)) {
|
|
auto j = nlohmann::json::parse(st->column_text(3), nullptr, false);
|
|
proto::Schedule sched;
|
|
if (auto p = proto::parse<proto::Schedule>(j, "schedule")) sched = *p;
|
|
q.window_open = window_open(sched, now);
|
|
}
|
|
queues.push_back(std::move(q));
|
|
}
|
|
}
|
|
|
|
// --- snapshot: tasks ----------------------------------------------------------
|
|
store::Tasks tasks(db_);
|
|
proto::TaskFilter filter;
|
|
filter.states = non_terminal_states();
|
|
auto page = tasks.list(filter, std::nullopt, 0, 100000);
|
|
if (!page) return std::unexpected(page.error());
|
|
|
|
std::vector<TaskView> views;
|
|
views.reserve(page->rows.size());
|
|
for (const auto& r : page->rows) {
|
|
if (r.state == "new") continue; // parked until the user starts it
|
|
TaskView v;
|
|
v.task_id = r.task_id;
|
|
v.run_state = run_state_of(r.state);
|
|
v.pause_reason = pause_reason_of(r.pause_reason);
|
|
v.queue_id = r.queue_id;
|
|
v.queue_position = r.queue_position.value_or(0);
|
|
v.host = host_of(r.url);
|
|
v.admit_rank = rank_of(r.created_at);
|
|
views.push_back(std::move(v));
|
|
}
|
|
|
|
const Decision d = governor_.evaluate(views, queues);
|
|
|
|
// --- apply ------------------------------------------------------------------
|
|
// to_start: probe first, always — a real ProbeResult (size, resumable, validator) is
|
|
// what makes sizeBytes/resumable correct on the wire, not a post-hoc guess. The task
|
|
// moves to `probing` immediately so the governor does not re-admit it on the next
|
|
// tick while the (possibly slow, always async) probe is outstanding.
|
|
for (const auto& wire_id : d.to_start) {
|
|
auto got = tasks.get(wire_id);
|
|
if (!got || !got->has_value()) continue;
|
|
const store::TaskRow& row = **got;
|
|
|
|
transition(wire_id, "probing", std::nullopt, std::nullopt);
|
|
|
|
vdm::net::ProbeRequest req;
|
|
req.url = row.url;
|
|
// headers / cookies / referrer / user_agent are not persisted yet (a URL-only
|
|
// `velox add` has none); the capture path will fill them when it lands.
|
|
|
|
const std::string id_copy = wire_id;
|
|
engine_.probe(req, [this, id_copy](vdm::Result<vdm::net::ProbeResult> pr) {
|
|
deps_.post_to_loop([this, id_copy, pr]() { on_probe_result(id_copy, pr); });
|
|
});
|
|
}
|
|
|
|
for (const auto& wire_id : d.to_resume) {
|
|
if (auto eid = engine_id_of(wire_id)) {
|
|
engine_.resume(*eid);
|
|
transition(wire_id, "connecting", std::nullopt, std::nullopt);
|
|
}
|
|
}
|
|
|
|
for (const auto& wire_id : d.to_pause) {
|
|
const auto reason = d.pause_reasons.count(wire_id)
|
|
? pause_reason_str(d.pause_reasons.at(wire_id))
|
|
: "user";
|
|
if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid);
|
|
transition(wire_id, "paused", std::string(reason), std::nullopt);
|
|
}
|
|
|
|
std::vector<vdm::TaskId> order;
|
|
order.reserve(d.priority_order.size());
|
|
for (const auto& wire_id : d.priority_order)
|
|
if (auto eid = engine_id_of(wire_id)) order.push_back(*eid);
|
|
engine_.set_task_order(order);
|
|
|
|
return {};
|
|
}
|
|
|
|
void Scheduler::transition(const std::string& wire_id, std::string_view to_state,
|
|
std::optional<std::string> pause_reason,
|
|
const std::optional<TaskErrorFields>& err) {
|
|
store::Tasks tasks(db_);
|
|
const auto before = tasks.get(wire_id);
|
|
const std::string previous = (before && before->has_value()) ? (**before).state : std::string();
|
|
|
|
// An engine-initiated pause carries an error => 'auto' (ADR 0013 §2), overriding
|
|
// whatever the caller passed (a scheduler-driven pause never carries an error here).
|
|
std::optional<std::string> reason = pause_reason;
|
|
if (to_state == "paused" && err) {
|
|
reason = "auto";
|
|
} else if (to_state == "paused" && !reason && before && before->has_value()) {
|
|
// No reason supplied — the common case is the engine's own pause-ack callback
|
|
// (on_state(_, paused, nullopt)) arriving after whoever actually initiated the
|
|
// pause (user_pause() or tick()'s to_pause loop) already wrote the real reason
|
|
// eagerly. Keep what's already stored instead of clobbering it back to NULL:
|
|
// set_state() always overwrites the column, reason or not.
|
|
reason = (**before).pause_reason;
|
|
}
|
|
(void)tasks.set_state(wire_id, to_state, reason);
|
|
|
|
if (err) {
|
|
auto st = db_.prepare(
|
|
"UPDATE tasks SET error_code=?2, error_message=?3, error_http_status=?4, "
|
|
"error_retryable=?5 WHERE task_id=?1");
|
|
if (st) {
|
|
(void)st->bind(1, std::string_view(wire_id));
|
|
(void)st->bind(2, std::string_view(err->code));
|
|
(void)st->bind(3, std::string_view(err->message));
|
|
if (err->http_status) (void)st->bind(4, *err->http_status);
|
|
else (void)st->bind_null(4);
|
|
if (err->retryable) (void)st->bind(5, static_cast<std::int64_t>(*err->retryable));
|
|
else (void)st->bind_null(5);
|
|
(void)st->step();
|
|
}
|
|
}
|
|
|
|
if (!hub_) return;
|
|
auto after = tasks.get(wire_id);
|
|
if (!after || !after->has_value()) return;
|
|
const proto::TaskSummary summary = store::to_summary(**after);
|
|
|
|
nlohmann::json params{
|
|
{"taskId", wire_id},
|
|
{"state", std::string(to_state)},
|
|
{"previousState", previous.empty() ? nlohmann::json(nullptr) : nlohmann::json(previous)},
|
|
{"summary", summary},
|
|
{"error", summary.error.has_value() ? nlohmann::json(*summary.error) : nlohmann::json(nullptr)},
|
|
};
|
|
hub_->publish(proto::Event::TaskState, proto::make_notification(proto::Event::TaskState, params),
|
|
wire_id);
|
|
}
|
|
|
|
void Scheduler::on_engine_state(const std::string& wire_id, std::string_view from_state,
|
|
std::string_view to_state,
|
|
const std::optional<TaskErrorFields>& err) {
|
|
(void)from_state; // transition() reads the store's own current state as previousState,
|
|
// which is authoritative regardless of engine/store timing
|
|
transition(wire_id, to_state, std::nullopt, err);
|
|
|
|
if (to_state == "complete" || to_state == "failed" || to_state == "cancelled") {
|
|
if (auto eid = engine_id_of(wire_id)) {
|
|
// One last snapshot before the handle goes away: a task that never lived past
|
|
// a single tick (small/fast/local) would otherwise leave downloadedBytes and
|
|
// segmentDetail at their pre-segmentation defaults forever, in violation of
|
|
// TaskDetail.segmentDetail's "exactly summary.segments entries" contract.
|
|
if (const auto p = engine_.progress(*eid)) persist_progress(wire_id, *p);
|
|
engine_.release(*eid);
|
|
unmap_engine(*eid);
|
|
}
|
|
}
|
|
}
|
|
|
|
vdm::task::DownloadCallbacks Scheduler::make_callbacks(const std::string& wire_id) {
|
|
vdm::task::DownloadCallbacks cbs;
|
|
const std::string id_copy = wire_id;
|
|
|
|
cbs.on_state = [this, id_copy](vdm::task::EngineState from, vdm::task::EngineState to,
|
|
const std::optional<vdm::ErrorInfo>& err) {
|
|
std::optional<TaskErrorFields> ef;
|
|
if (err) ef = to_error_fields(*err);
|
|
const std::string from_name = engine_state_name(from);
|
|
const std::string to_name = engine_state_name(to);
|
|
deps_.post_to_loop([this, id_copy, from_name, to_name, ef]() {
|
|
on_engine_state(id_copy, from_name, to_name, ef);
|
|
});
|
|
};
|
|
|
|
cbs.on_finished = [this, id_copy](vdm::Result<vdm::task::DownloadOutcome> outcome) {
|
|
deps_.post_to_loop([this, id_copy, outcome]() { on_engine_finished(id_copy, outcome); });
|
|
};
|
|
|
|
return cbs;
|
|
}
|
|
|
|
void Scheduler::on_probe_result(const std::string& wire_id,
|
|
const vdm::Result<vdm::net::ProbeResult>& pr) {
|
|
auto got = store::Tasks(db_).get(wire_id);
|
|
if (!got || !got->has_value()) return; // removed while the probe was outstanding
|
|
const store::TaskRow& row = **got;
|
|
|
|
if (!pr) {
|
|
transition(wire_id, "failed", std::nullopt, to_error_fields(pr.error()));
|
|
return;
|
|
}
|
|
|
|
store::Tasks::ProbeFields fields;
|
|
if (pr->total_size) fields.size_bytes = static_cast<std::int64_t>(*pr->total_size);
|
|
fields.resumable = pr->resumable;
|
|
if (!pr->etag.empty()) fields.etag = pr->etag;
|
|
if (!pr->last_modified.empty()) fields.last_modified = pr->last_modified;
|
|
if (!pr->mime.empty()) fields.content_type = pr->mime;
|
|
if (pr->effective_url != row.url) fields.effective_url = pr->effective_url;
|
|
(void)store::Tasks(db_).set_probe_result(wire_id, fields);
|
|
|
|
vdm::task::DownloadSpec spec;
|
|
spec.url = row.url;
|
|
spec.save_path = row.save_dir + "/" + row.filename;
|
|
if (row.req_segments) spec.segments = static_cast<std::uint32_t>(*row.req_segments);
|
|
if (row.req_buffer_bytes) spec.buffer_bytes = static_cast<std::uint64_t>(*row.req_buffer_bytes);
|
|
if (row.checksum_algo && row.checksum_value) {
|
|
vdm::task::Checksum ck;
|
|
ck.hex = *row.checksum_value;
|
|
if (*row.checksum_algo == "md5") ck.algo = vdm::task::Checksum::Algo::md5;
|
|
else if (*row.checksum_algo == "sha1") ck.algo = vdm::task::Checksum::Algo::sha1;
|
|
else if (*row.checksum_algo == "sha512") ck.algo = vdm::task::Checksum::Algo::sha512;
|
|
else ck.algo = vdm::task::Checksum::Algo::sha256;
|
|
spec.checksum = ck;
|
|
}
|
|
spec.allow_resume = true; // resume from a sidecar if one is beside save_path
|
|
spec.probe_hint = *pr; // skip a second probe; the engine still revalidates on resume
|
|
|
|
const vdm::TaskId engine_id = engine_.start(spec, make_callbacks(wire_id));
|
|
map(wire_id, engine_id);
|
|
// State stays `probing`; the engine's own on_state (probe_hint => starts in
|
|
// `connecting`) drives the next transition through on_engine_state.
|
|
}
|
|
|
|
void Scheduler::on_engine_finished(const std::string& wire_id,
|
|
const vdm::Result<vdm::task::DownloadOutcome>& outcome) {
|
|
// The state transition (complete/failed/cancelled) already happened via on_state,
|
|
// which always precedes on_finished. This only tops up the byte count for a task that
|
|
// completed before any progress tick ran — otherwise a fast/local/small transfer
|
|
// reports downloadedBytes: 0 forever despite a byte-correct file on disk.
|
|
if (outcome) (void)store::Tasks(db_).set_final_bytes(wire_id, static_cast<std::int64_t>(outcome->bytes));
|
|
}
|
|
|
|
void Scheduler::persist_progress(const std::string& wire_id, const vdm::task::Progress& p) {
|
|
// TaskDetail.segmentDetail is contractually "exactly TaskSummary.segments entries" —
|
|
// so the count that goes on the wire as `segments` has to be the length of the list
|
|
// that actually becomes segmentDetail, not effective_segments (budget slots *held*,
|
|
// per engine_port.hpp; a small file can hold 8 fairness slots while its segmenter
|
|
// only ever carves 2 ranges). Falls back to effective_segments only before the task
|
|
// has any ranges yet, so a `probing`/`connecting` task still reports a sane count.
|
|
const std::int64_t seg_count = !p.segments.empty()
|
|
? static_cast<std::int64_t>(p.segments.size())
|
|
: static_cast<std::int64_t>(p.effective_segments);
|
|
|
|
(void)store::Tasks(db_).update_progress(wire_id, static_cast<std::int64_t>(p.downloaded),
|
|
static_cast<std::int64_t>(p.speed_bps), seg_count,
|
|
static_cast<std::int64_t>(p.effective_buffer_bytes));
|
|
|
|
if (!p.segments.empty()) {
|
|
std::vector<store::SegmentSnapshot> snaps;
|
|
snaps.reserve(p.segments.size());
|
|
for (const auto& s : p.segments) {
|
|
snaps.push_back({s.index, static_cast<std::int64_t>(s.start),
|
|
static_cast<std::int64_t>(s.end),
|
|
static_cast<std::int64_t>(s.completed),
|
|
static_cast<std::int64_t>(s.speed_bps),
|
|
segment_state_name(s.state)});
|
|
}
|
|
(void)store::Segments(db_).replace_all(wire_id, snaps);
|
|
}
|
|
}
|
|
|
|
std::vector<Scheduler::ProgressRow> Scheduler::progress_snapshot() {
|
|
std::vector<ProgressRow> out;
|
|
if (to_engine_.empty()) return out;
|
|
|
|
out.reserve(to_engine_.size());
|
|
for (const auto& [wire_id, engine_id] : to_engine_) {
|
|
const auto p = engine_.progress(engine_id);
|
|
if (!p) continue;
|
|
|
|
ProgressRow row;
|
|
row.task_id = wire_id;
|
|
row.downloaded_bytes = p->downloaded;
|
|
row.speed_bps = p->speed_bps;
|
|
row.eta_seconds = p->eta_seconds;
|
|
for (const auto& s : p->segments)
|
|
row.segments.push_back({s.index, s.completed, s.speed_bps});
|
|
out.push_back(std::move(row));
|
|
|
|
persist_progress(wire_id, *p);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
namespace {
|
|
bool is_terminal_state(const std::string& s) {
|
|
return s == "complete" || s == "failed" || s == "cancelled";
|
|
}
|
|
} // namespace
|
|
|
|
rpc::TaskActionPort::Result Scheduler::user_pause(const std::string& wire_id) {
|
|
store::Tasks tasks(db_);
|
|
auto got = tasks.get(wire_id);
|
|
if (!got || !got->has_value()) return {false, false, {}};
|
|
const store::TaskRow& row = **got;
|
|
if (is_terminal_state(row.state) || row.state == "paused")
|
|
return {true, false, row.state};
|
|
|
|
if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid);
|
|
transition(wire_id, "paused", std::string("user"), std::nullopt);
|
|
return {true, true, "paused"};
|
|
}
|
|
|
|
rpc::TaskActionPort::Result Scheduler::user_resume(const std::string& wire_id) {
|
|
store::Tasks tasks(db_);
|
|
auto got = tasks.get(wire_id);
|
|
if (!got || !got->has_value()) return {false, false, {}};
|
|
const store::TaskRow& row = **got;
|
|
if (row.state != "paused") return {true, false, row.state};
|
|
|
|
if (auto eid = engine_id_of(wire_id)) {
|
|
engine_.resume(*eid);
|
|
transition(wire_id, "connecting", std::nullopt, std::nullopt);
|
|
return {true, true, "connecting"};
|
|
}
|
|
transition(wire_id, "queued", std::nullopt, std::nullopt);
|
|
return {true, true, "queued"};
|
|
}
|
|
|
|
rpc::TaskActionPort::Result Scheduler::user_start(const std::string& wire_id) {
|
|
store::Tasks tasks(db_);
|
|
auto got = tasks.get(wire_id);
|
|
if (!got || !got->has_value()) return {false, false, {}};
|
|
const store::TaskRow& row = **got;
|
|
if (row.state != "paused" && row.state != "new") return {true, false, row.state};
|
|
|
|
if (auto eid = engine_id_of(wire_id)) {
|
|
engine_.resume(*eid);
|
|
transition(wire_id, "connecting", std::nullopt, std::nullopt);
|
|
return {true, true, "connecting"};
|
|
}
|
|
transition(wire_id, "queued", std::nullopt, std::nullopt);
|
|
return {true, true, "queued"};
|
|
}
|
|
|
|
rpc::TaskActionPort::Result Scheduler::user_cancel(const std::string& wire_id,
|
|
bool discard_partial) {
|
|
store::Tasks tasks(db_);
|
|
auto got = tasks.get(wire_id);
|
|
if (!got || !got->has_value()) return {false, false, {}};
|
|
const store::TaskRow& row = **got;
|
|
if (is_terminal_state(row.state)) return {true, false, row.state};
|
|
|
|
// Same pattern as tick()'s to_pause loop: call the engine (async, no synchronous
|
|
// effect) and transition the store eagerly so download.get/list are correct the
|
|
// instant this call returns. The engine's own on_state(_, cancelled, nullopt) +
|
|
// on_finished arrive later via on_engine_state, which is what actually
|
|
// release()s/unmaps the handle — never done here.
|
|
if (auto eid = engine_id_of(wire_id)) engine_.cancel(*eid, discard_partial);
|
|
transition(wire_id, "cancelled", std::nullopt, std::nullopt);
|
|
return {true, true, "cancelled"};
|
|
}
|
|
|
|
std::vector<std::string> Scheduler::pause_queue(const std::string& queue_id) {
|
|
store::Tasks tasks(db_);
|
|
proto::TaskFilter filter;
|
|
filter.queueId = queue_id;
|
|
filter.states = non_terminal_states();
|
|
// No paging needed: a queue's max_concurrent is <= 32, so "everything non-terminal in
|
|
// this queue" is never a large page.
|
|
auto page = tasks.list(filter, std::nullopt, 0, 10000);
|
|
std::vector<std::string> paused;
|
|
if (!page) return paused;
|
|
|
|
for (const auto& row : page->rows) {
|
|
if (run_state_of(row.state) != RunState::Running) continue;
|
|
if (auto eid = engine_id_of(row.task_id)) engine_.pause(*eid);
|
|
transition(row.task_id, "paused", std::string("queue_stopped"), std::nullopt);
|
|
paused.push_back(row.task_id);
|
|
}
|
|
return paused;
|
|
}
|
|
|
|
bool Scheduler::provide_auth(const std::string& wire_id, const std::string& username,
|
|
const std::string& password, bool remember) {
|
|
(void)remember; // not yet wired to the Secret Service anywhere in this build
|
|
auto eid = engine_id_of(wire_id);
|
|
if (!eid) return false;
|
|
engine_.provide_auth(*eid, username, password, remember);
|
|
return true;
|
|
}
|
|
|
|
|
|
void Scheduler::probe_now(
|
|
const proto::DownloadProbeParams& params,
|
|
std::function<void(proto::HandlerResult<proto::DownloadProbeResult>)> 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<vdm::net::ProbeResult> 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<std::int64_t>(*pr->total_size);
|
|
r.mime = pr->mime;
|
|
r.resumable = pr->resumable;
|
|
r.effectiveUrl = pr->effective_url.empty() ? params.url : pr->effective_url;
|
|
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;
|
|
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);
|
|
});
|
|
});
|
|
}
|
|
|
|
void Scheduler::refresh_url(
|
|
const std::string& wire_id, const std::string& url,
|
|
const std::optional<proto::Headers>& headers,
|
|
const std::optional<std::vector<proto::Cookie>>& cookies,
|
|
std::function<void(proto::HandlerResult<proto::DownloadRefreshUrlResult>)> done) {
|
|
auto got = store::Tasks(db_).get(wire_id);
|
|
if (!got || !got->has_value()) {
|
|
done(std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task",
|
|
nlohmann::json{{"taskId", wire_id}}}));
|
|
return;
|
|
}
|
|
const store::TaskRow row = **got; // copied: read again by the async callback below
|
|
|
|
vdm::net::ProbeRequest req;
|
|
req.url = url;
|
|
if (headers)
|
|
for (const auto& [k, v] : *headers) req.headers.push_back({k, v});
|
|
if (cookies)
|
|
for (const auto& c : *cookies) req.cookies.push_back({c.name, c.value});
|
|
|
|
engine_.probe(req, [this, wire_id, row, url, headers, done](vdm::Result<vdm::net::ProbeResult> pr) {
|
|
deps_.post_to_loop([this, wire_id, row, url, headers, 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;
|
|
}
|
|
|
|
// "if they do not [match], it says so rather than silently restarting" (the
|
|
// schema's own words) — comparison only fires when both sides actually have a
|
|
// value; an unknown size/validator on either end is not itself a mismatch.
|
|
bool content_changed = false;
|
|
if (row.size_bytes && pr->total_size &&
|
|
*row.size_bytes != static_cast<std::int64_t>(*pr->total_size))
|
|
content_changed = true;
|
|
if (row.etag && !row.etag->empty() && !pr->etag.empty() && *row.etag != pr->etag)
|
|
content_changed = true;
|
|
if (row.last_modified && !row.last_modified->empty() && !pr->last_modified.empty() &&
|
|
*row.last_modified != pr->last_modified)
|
|
content_changed = true;
|
|
|
|
store::Tasks tasks(db_);
|
|
store::Tasks::ProbeFields fields;
|
|
if (pr->total_size) fields.size_bytes = static_cast<std::int64_t>(*pr->total_size);
|
|
fields.resumable = pr->resumable;
|
|
if (!pr->etag.empty()) fields.etag = pr->etag;
|
|
if (!pr->last_modified.empty()) fields.last_modified = pr->last_modified;
|
|
if (!pr->mime.empty()) fields.content_type = pr->mime;
|
|
const std::string effective = pr->effective_url.empty() ? url : pr->effective_url;
|
|
fields.effective_url = effective;
|
|
(void)tasks.set_probe_result(wire_id, fields);
|
|
(void)tasks.set_url(wire_id, url);
|
|
|
|
if (auto eid = engine_id_of(wire_id)) {
|
|
std::vector<vdm::net::HeaderField> hdrs;
|
|
if (headers)
|
|
for (const auto& [k, v] : *headers) hdrs.push_back({k, v});
|
|
engine_.refresh_url(*eid, url, hdrs);
|
|
}
|
|
|
|
proto::DownloadRefreshUrlResult r;
|
|
r.ok = true;
|
|
r.resumable = pr->resumable;
|
|
r.contentChanged = content_changed;
|
|
if (pr->total_size) r.sizeBytes = static_cast<std::int64_t>(*pr->total_size);
|
|
r.effectiveUrl = effective;
|
|
done(r);
|
|
});
|
|
});
|
|
}
|
|
|
|
} // namespace velox::daemon::sched
|