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

daemon: persist engine progress/probe to the store — segments:0 no longer leaks

Browse Source 
Engine numbers never reached the store: download.get after a correctly-finished
download reported sizeBytes: null, downloadedBytes: 0, speedBps: 0, resumable:
false, segments: 0, segmentDetail: [] — segments: 0 breaks the frozen contract
(TaskSummary.segments is minimum:1, required).

- Scheduler::tick() now probes (EnginePort::probe) before every start(), persisting
  sizeBytes/resumable/validators via Tasks::set_probe_result before a byte moves,
  then starts the engine with that ProbeResult as probe_hint.
- Scheduler::persist_progress() (new) writes downloadedBytes/speedBps/segments/
  segmentDetail from the engine's Progress. Called from progress_snapshot() (the
  250ms tick) *and* once more from on_engine_state right before release()/unmap on
  every terminal transition, so a task that finishes between two ticks — the common
  case for anything small or fast — still leaves real numbers instead of the
  pre-persistence defaults.
- TaskSummary.segments is sourced from segments.size() when the task has any
  (matching what actually lands in segmentDetail, per the schema's "exactly
  segments entries"), falling back to the engine's effective_segments (budget
  slots *held*, not necessarily physical range count) only pre-segmentation.
- Tasks::set_final_bytes tops up on_finished's byte count as a last-resort
  backstop.
- store/segments.{cpp,hpp}: read/write access to the segments table behind
  TaskDetail.segmentDetail. Wired into daemon/CMakeLists.txt.
- migrations/0002: speed_bps on tasks and segments; fixes segments.state's CHECK
  to include 'pending' (0001 omitted it, so a pre-connect snapshot could never be
  written).
- store_migrations_test's forward-only loop faked "released version N" by setting
  the user_version pragma alone, with no real schema underneath — never exercised
  until 0002 existed. Fixed to actually build the db through migrations 1..N first.

Verified against real veloxd + tools/testserver (not just unit tests):
download.list/download.get correct immediately after completion and after a
daemon restart, with saveTo.allowedRoots pointed at an isolated dir.

Observed but not fixed (CORE, not this lane, noted in deferrals.md): Progress.
speed_bps reads back 0 for the whole lifetime of a live throttled download in the
same E2E check, despite downloadedBytes visibly advancing. DAEMON passes it
through unmodified; filed rather than worked around.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP
This commit is contained in:
sami
2026-09-11 12:11:19 +04:00
co-authored by Claude Sonnet 5
parent 824fa481bb
commit a967eca669
14 changed files with 448 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/src/sched/scheduler.cpp
+143 -41
View File
@@ -6,6 +6,7 @@
#include <nlohmann/json.hpp>
#include "sched/schedule_window.hpp"
#include "store/segments.hpp"
#include "store/settings.hpp"
#include "store/tasks.hpp"
@@ -100,6 +101,28 @@ std::vector<proto::TaskState> non_terminal_states() {
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,
@@ -212,52 +235,26 @@ store::DbResult<void> Scheduler::tick() {
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;
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
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.
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 = TaskErrorFields{};
ef->code = std::string(vdm::error_name(err->code)); // matches TaskErrorCode
ef->message = err->context;
if (err->http_status != 0) ef->http_status = err->http_status;
ef->retryable = err->retryable;
}
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);
});
};
const vdm::TaskId engine_id = engine_.start(spec, std::move(cbs));
map(wire_id, engine_id);
transition(wire_id, "probing", std::nullopt, std::nullopt);
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) {
@@ -338,17 +335,124 @@ void Scheduler::on_engine_state(const std::string& wire_id, std::string_view fro
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;
store::Tasks tasks(db_);
out.reserve(to_engine_.size());
for (const auto& [wire_id, engine_id] : to_engine_) {
const auto p = engine_.progress(engine_id);
@@ -363,9 +467,7 @@ std::vector<Scheduler::ProgressRow> Scheduler::progress_snapshot() {
row.segments.push_back({s.index, s.completed, s.speed_bps});
out.push_back(std::move(row));
(void)tasks.update_progress(wire_id, static_cast<std::int64_t>(p->downloaded),
static_cast<std::int64_t>(p->effective_segments),
static_cast<std::int64_t>(p->effective_buffer_bytes));
persist_progress(wire_id, *p);
}
return out;
}