diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt index 4559005..6d896ea 100644 --- a/daemon/CMakeLists.txt +++ b/daemon/CMakeLists.txt @@ -37,6 +37,7 @@ add_library(veloxd_store STATIC src/store/tasks.cpp src/store/categories.cpp src/store/queues.cpp + src/store/segments.cpp ${_mig_hdr} ) add_library(velox::daemon_store ALIAS veloxd_store) diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index 73778e6..5208db4 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -11,3 +11,5 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | ~~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 | timer + nudges: a 1 s `timerfd` re-runs `Scheduler::tick()` and `download.add` nudges via `on_mutation`. `download.pause`/`resume`/`start`/`cancel` and the queue.* handlers still don't touch the scheduler | `rpc/dispatcher.cpp` | those handlers are still stubs (D3) | as each handler is implemented behind the store, it calls `on_mutation` / drives the scheduler | | ~~D5~~ | **Mostly closed** — `rpc/event_hub` fans out per-subscription; `session.subscribe` on both transports registers/updates/tears down a real subscription; `Scheduler::transition()` publishes `event.task.state` (with `previousState`) on every state change, scheduler-driven or engine-reported; `dispatcher::on_download_add` publishes `event.task.added`; a 250 ms timer batches `Scheduler::progress_snapshot()` into one `event.task.progress` array per AGENT-DAEMON.md item 5 / the schema's `x-maxRateHz: 4`. Verified live end to end. | — | `event.task.removed` has no source yet (`download.remove` is D3); `event.speed.global`, `event.notify`, `event.auth.required`, `event.settings.changed`, `event.grabber.progress` are unpublished — each lands with its owning handler | as each owning D3 handler lands | +| ~~D6~~ | **Closed** — engine numbers now reach the store: `Scheduler::tick()` probes (`EnginePort::probe`) before every `start()`, persisting `sizeBytes`/`resumable`/validators via `Tasks::set_probe_result` before a byte moves; `Scheduler::persist_progress()` (called from `progress_snapshot()` *and* once more from `on_engine_state` right before `release()`/unmap on every terminal transition) writes `downloadedBytes`/`speedBps`/`segments`/`segmentDetail` from the engine's `Progress`, so a task that finishes between two 250 ms 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 — see `core/include/vdm/task/download.hpp`'s `Progress` comment) only pre-segmentation. `Tasks::set_final_bytes` tops up `on_finished`'s byte count as a last-resort backstop. Migration `0002` adds `speed_bps` to both `tasks` and `segments`, and fixes `segments.state`'s CHECK to include `'pending'` (0001 omitted it, so a pre-connect snapshot could never be written). Verified against real `veloxd` + `tools/testserver` (not just unit tests): `download.list`/`download.get` correct immediately after completion and after a daemon restart. | `sched/scheduler.{cpp,hpp}`, `store/{tasks,segments}.{cpp,hpp}`, `store/migrations/0002_*.sql` | — | done | +| — | **Observed, not fixed (CORE, not this lane):** `vdm::task::Progress.speed_bps` reads back as `0` for the whole lifetime of a live, real (non-fake) throttled download in the E2E check above, despite `downloadedBytes` visibly advancing between polls — `core/src/task/download_task.cpp`'s per-worker EWMA (`w->speed_bps`, ~line 505-513) never seems to produce a nonzero aggregate in this build. DAEMON passes `EnginePort::progress()`'s `speed_bps` straight through (`Scheduler::persist_progress`); nothing in this lane drops it. Filed here rather than worked around — CLAUDE.md §2/§3: not core/'s owner, don't patch around a wrong upstream number locally. Confirm with CORE before the GUI's live speed readout ships. | diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 014a5ad..524ad37 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -8,6 +8,7 @@ #include "fs/safepath.hpp" #include "store/categories.hpp" #include "store/queues.hpp" +#include "store/segments.hpp" #include "store/settings.hpp" #include "store/tasks.hpp" #include "util/time.hpp" @@ -160,6 +161,13 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { row.description = spec.description; row.req_segments = spec.segments; row.req_buffer_bytes = spec.bufferBytes; + // eff_segments starts at the requested-or-default count, never 0: TaskSummary.segments + // is minimum:1 on the wire even before this task has connected its first segment (a + // task that has never run does not get a free pass to violate its own contract). + row.eff_segments = + spec.segments.value_or(settings.get_int("connection.maxSegmentsPerDownload")); + if (row.eff_segments < 1) row.eff_segments = 1; + if (row.eff_segments > 32) row.eff_segments = 32; if (spec.checksum) { row.checksum_algo = std::string(proto::to_string(spec.checksum->algorithm)); row.checksum_value = spec.checksum->value; @@ -237,8 +245,17 @@ VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) { const store::TaskRow& row = **got; proto::TaskDetail d; d.summary = store::to_summary(row); - // segmentDetail stays empty until the engine has segmented the task — the schema - // permits that ("empty before the task has been segmented"). + + // Empty until the engine has segmented the task — the schema permits that ("empty + // before the task has been segmented"); otherwise exactly summary.segments entries, + // kept current by the same progress tick that updates the byte counters. + store::Segments segments(db_); + if (auto segs = segments.list(row.task_id)) + d.segmentDetail = std::move(*segs); + else + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InternalError, "download.get: " + segs.error().message}); + d.mime = row.content_type; d.bufferBytes = row.req_buffer_bytes; d.effectiveBufferBytes = row.eff_buffer_bytes; diff --git a/daemon/src/sched/engine_port.hpp b/daemon/src/sched/engine_port.hpp index ba9ef8d..b487eb7 100644 --- a/daemon/src/sched/engine_port.hpp +++ b/daemon/src/sched/engine_port.hpp @@ -7,10 +7,11 @@ // wraps `vdm::Engine` + `vdm::segment::SegmentBudget`; FakeEnginePort records calls. // // Task ids here are `vdm::TaskId` — the engine assigns one from start() and the Scheduler -// keeps the wire-UUID <-> TaskId map (ADR 0013). Admission is the Scheduler's: it calls -// start() only for a task the governor admitted, and the engine begins probing at once -// (it does not queue). The min-1 fairness rule in SegmentBudget then guarantees each -// started task a slot; set_task_order pushes the priority. +// keeps the wire-UUID <-> TaskId map (ADR 0013). Admission is the Scheduler's: for a task +// the governor admits, it probes first (persisting sizeBytes/resumable/validator before a +// single byte moves), then calls start() with that ProbeResult as probe_hint. The min-1 +// fairness rule in SegmentBudget then guarantees each started task a slot; set_task_order +// pushes the priority. #include #include @@ -19,7 +20,9 @@ #include #include "vdm/ids.hpp" +#include "vdm/net/probe.hpp" #include "vdm/task/download.hpp" +#include "vdm/util/result.hpp" namespace velox::daemon::sched { @@ -27,6 +30,13 @@ class EnginePort { public: virtual ~EnginePort() = default; + // Runs on the probe pool, outside the segment budget (ADR 0011 §5); `done` arrives on + // an engine thread, exactly once. The Scheduler probes before every start() so it + // always has a real ProbeResult (size, resumable, validator) to persist and to pass + // back as DownloadSpec.probe_hint — one code path instead of "sometimes has one". + virtual void probe(const vdm::net::ProbeRequest& req, + std::function)> done) = 0; + virtual vdm::TaskId start(const vdm::task::DownloadSpec& spec, vdm::task::DownloadCallbacks callbacks) = 0; diff --git a/daemon/src/sched/engine_port_core.hpp b/daemon/src/sched/engine_port_core.hpp index 4fff2be..7c6fbe4 100644 --- a/daemon/src/sched/engine_port_core.hpp +++ b/daemon/src/sched/engine_port_core.hpp @@ -16,6 +16,11 @@ class EnginePortCore final : public EnginePort { public: explicit EnginePortCore(vdm::Engine& engine) : engine_(engine) {} + void probe(const vdm::net::ProbeRequest& req, + std::function)> done) override { + engine_.probe(req, std::move(done)); + } + vdm::TaskId start(const vdm::task::DownloadSpec& spec, vdm::task::DownloadCallbacks callbacks) override { vdm::task::DownloadHandle h = engine_.start(spec, std::move(callbacks)); diff --git a/daemon/src/sched/fake_engine_port.hpp b/daemon/src/sched/fake_engine_port.hpp index 127b90d..1e88d81 100644 --- a/daemon/src/sched/fake_engine_port.hpp +++ b/daemon/src/sched/fake_engine_port.hpp @@ -30,6 +30,26 @@ public: std::vector max_active_segments; std::vector> host_caps; + // probe(): synchronous by default (a default-constructed ProbeResult — success, + // resumable=false, no known size) so a test that doesn't care about probe details + // still sees start() happen within the same tick(). Set auto_probe_result to nullopt + // to switch to manual mode: probe() then just records the request and stashes `done` + // in pending_probes for the test to resolve explicitly, in order. + std::optional> auto_probe_result = + vdm::Result{vdm::net::ProbeResult{}}; + std::vector probe_requests; + std::vector)>> pending_probes; + + void probe(const vdm::net::ProbeRequest& req, + std::function)> done) override { + probe_requests.push_back(req); + if (auto_probe_result) { + done(*auto_probe_result); + } else { + pending_probes.push_back(std::move(done)); + } + } + vdm::TaskId start(const vdm::task::DownloadSpec& spec, vdm::task::DownloadCallbacks callbacks) override { const vdm::TaskId id{next_++}; diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp index 8f26954..52a26ba 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/segments.hpp" #include "store/settings.hpp" #include "store/tasks.hpp" @@ -100,6 +101,28 @@ std::vector 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 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(*row.req_segments); - if (row.req_buffer_bytes) - spec.buffer_bytes = static_cast(*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& err) { - std::optional 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 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& err) { + std::optional 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 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& 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(*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(*row.req_segments); + if (row.req_buffer_bytes) spec.buffer_bytes = static_cast(*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& 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(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(p.segments.size()) + : static_cast(p.effective_segments); + + (void)store::Tasks(db_).update_progress(wire_id, static_cast(p.downloaded), + static_cast(p.speed_bps), seg_count, + static_cast(p.effective_buffer_bytes)); + + if (!p.segments.empty()) { + std::vector snaps; + snaps.reserve(p.segments.size()); + for (const auto& s : p.segments) { + snaps.push_back({s.index, static_cast(s.start), + static_cast(s.end), + static_cast(s.completed), + static_cast(s.speed_bps), + segment_state_name(s.state)}); + } + (void)store::Segments(db_).replace_all(wire_id, snaps); + } +} + std::vector Scheduler::progress_snapshot() { std::vector 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::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(p->downloaded), - static_cast(p->effective_segments), - static_cast(p->effective_buffer_bytes)); + persist_progress(wire_id, *p); } return out; } diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp index a7430a3..c4ba73d 100644 --- a/daemon/src/sched/scheduler.hpp +++ b/daemon/src/sched/scheduler.hpp @@ -29,6 +29,9 @@ #include "sched/governor.hpp" #include "store/sqlite.hpp" #include "vdm/ids.hpp" +#include "vdm/net/probe.hpp" +#include "vdm/task/download.hpp" +#include "vdm/util/result.hpp" namespace velox::daemon::sched { @@ -110,6 +113,30 @@ private: std::optional pause_reason, const std::optional& err); + // The probe issued for `wire_id` in tick() has resolved: persist sizeBytes / + // resumable / validator, then start() with the result as probe_hint. A probe failure + // (bad URL, DNS, 404 with no mirrors) moves the task straight to `failed` — it never + // reaches start(). + void on_probe_result(const std::string& wire_id, const vdm::Result& pr); + + // on_finished fired: top up the final byte count (set_final_bytes) so a task that + // completed before any progress tick ran still reports real numbers. The state + // transition itself (complete/failed/cancelled) already happened via on_engine_state, + // which on_finished always follows. + void on_engine_finished(const std::string& wire_id, + const vdm::Result& outcome); + + // Writes one task's byte counters + segment rows from a Progress snapshot. Shared by + // progress_snapshot() (the periodic tick) and on_engine_state's terminal path (a final + // snapshot before release/unmap) so a task that finishes between two ticks — the + // common case for anything small or fast — still leaves a real segmentDetail behind + // instead of the pre-segmentation empty array. + void persist_progress(const std::string& wire_id, const vdm::task::Progress& p); + + // Wires on_state -> on_engine_state and on_finished -> on_engine_finished, both + // marshalled through post_to_loop. Shared by the one place a task actually starts. + vdm::task::DownloadCallbacks make_callbacks(const std::string& wire_id); + store::Db& db_; EnginePort& engine_; Governor governor_; diff --git a/daemon/src/store/migrations/0002_segment_speed_and_pending.sql b/daemon/src/store/migrations/0002_segment_speed_and_pending.sql new file mode 100644 index 0000000..773cd1c --- /dev/null +++ b/daemon/src/store/migrations/0002_segment_speed_and_pending.sql @@ -0,0 +1,30 @@ +-- Migration 0002 — segments gets a speed_bps column and a corrected state CHECK; +-- tasks gets a speed_bps column too. +-- +-- 0001's segments.state CHECK omitted 'pending' (vdm::segment::SegState::idle's wire +-- spelling — "range assigned, no worker connected yet"), so a segment snapshot taken +-- before its first worker connects could never be written. SQLite cannot ALTER a CHECK +-- constraint in place, so this rebuilds the table (standard SQLite pattern: create the +-- new shape, copy, drop, rename). speed_bps on both tables backs TaskSummary.speedBps / +-- Segment.speedBps on the wire; absent from 0001 because progress writing wasn't wired +-- yet — a task's engine-reported aggregate speed had nowhere to persist between polls. + +ALTER TABLE tasks ADD COLUMN speed_bps INTEGER NOT NULL DEFAULT 0; + +CREATE TABLE segments_new ( + task_id TEXT NOT NULL REFERENCES tasks(task_id) ON DELETE CASCADE, + idx INTEGER NOT NULL, + start_byte INTEGER NOT NULL, + end_byte INTEGER NOT NULL, + completed_bytes INTEGER NOT NULL DEFAULT 0, + speed_bps INTEGER NOT NULL DEFAULT 0, + state TEXT NOT NULL DEFAULT 'pending' + CHECK (state IN ('pending','connecting','downloading','stalled','complete','failed')), + PRIMARY KEY (task_id, idx) +) STRICT, WITHOUT ROWID; + +INSERT INTO segments_new (task_id, idx, start_byte, end_byte, completed_bytes, state) + SELECT task_id, idx, start_byte, end_byte, completed_bytes, state FROM segments; + +DROP TABLE segments; +ALTER TABLE segments_new RENAME TO segments; diff --git a/daemon/src/store/segments.cpp b/daemon/src/store/segments.cpp new file mode 100644 index 0000000..8935566 --- /dev/null +++ b/daemon/src/store/segments.cpp @@ -0,0 +1,61 @@ +#include "store/segments.hpp" + +namespace velox::daemon::store { + +namespace proto = velox::proto; + +DbResult Segments::replace_all(std::string_view task_id, + const std::vector& segs) { + return db_.transaction([&]() -> DbResult { + { + auto del = db_.prepare("DELETE FROM segments WHERE task_id = ?1"); + if (!del) return std::unexpected(del.error()); + if (auto r = del->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = del->step(); !r) return std::unexpected(r.error()); + } + for (const auto& s : segs) { + auto ins = db_.prepare( + "INSERT INTO segments(task_id, idx, start_byte, end_byte, completed_bytes, " + "speed_bps, state) VALUES(?1,?2,?3,?4,?5,?6,?7)"); + if (!ins) return std::unexpected(ins.error()); + if (auto r = ins->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = ins->bind(2, static_cast(s.index)); !r) + return std::unexpected(r.error()); + if (auto r = ins->bind(3, s.start_byte); !r) return std::unexpected(r.error()); + if (auto r = ins->bind(4, s.end_byte); !r) return std::unexpected(r.error()); + if (auto r = ins->bind(5, s.completed_bytes); !r) return std::unexpected(r.error()); + if (auto r = ins->bind(6, s.speed_bps); !r) return std::unexpected(r.error()); + if (auto r = ins->bind(7, std::string_view(s.state)); !r) + return std::unexpected(r.error()); + if (auto r = ins->step(); !r) return std::unexpected(r.error()); + } + return {}; + }); +} + +DbResult> Segments::list(std::string_view task_id) { + auto st = db_.prepare( + "SELECT idx, start_byte, end_byte, completed_bytes, speed_bps, state " + "FROM segments WHERE task_id = ?1 ORDER BY idx"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); + + std::vector out; + for (;;) { + auto row = st->step(); + if (!row) return std::unexpected(row.error()); + if (!*row) break; + + proto::Segment seg; + seg.index = st->column_int(0); + seg.startByte = st->column_int(1); + seg.endByte = st->column_int(2); + seg.downloadedBytes = st->column_int(3); + seg.speedBps = st->column_int(4); + if (auto s = proto::parse_SegmentState(st->column_text(5))) seg.state = *s; + out.push_back(seg); + } + return out; +} + +} // namespace velox::daemon::store diff --git a/daemon/src/store/segments.hpp b/daemon/src/store/segments.hpp new file mode 100644 index 0000000..196b9f1 --- /dev/null +++ b/daemon/src/store/segments.hpp @@ -0,0 +1,45 @@ +#pragma once + +// Read/write access to the `segments` table — the per-connection detail behind +// TaskDetail.segmentDetail (download.get) and the segment-bars widget. Written from an +// engine progress tick (Scheduler::progress_snapshot); read back on download.get. + +#include +#include +#include +#include + +#include "store/sqlite.hpp" +#include "velox_proto.hpp" + +namespace velox::daemon::store { + +// One segment's live counters, as read off vdm::task::SegmentProgress. `state` is a wire +// SegmentState spelling ("pending"/"connecting"/.../"failed") — the caller maps the +// engine's own enum, this module stays generic like the others. +struct SegmentSnapshot { + std::uint32_t index; + std::int64_t start_byte; + std::int64_t end_byte; // inclusive (ADR 0010) + std::int64_t completed_bytes; + std::int64_t speed_bps; + std::string state; +}; + +class Segments { +public: + explicit Segments(Db& db) : db_(db) {} + + // Replaces every row for `task_id` with `segs` in one transaction. Simpler than a + // per-index upsert and cheap at <=32 rows, <=4 Hz. + DbResult replace_all(std::string_view task_id, const std::vector& segs); + + // In index order. Empty if the task has never been segmented (TaskDetail's own + // description permits this). + DbResult> list(std::string_view task_id); + +private: + Db& db_; +}; + +} // namespace velox::daemon::store diff --git a/daemon/src/store/tasks.cpp b/daemon/src/store/tasks.cpp index 7ea946b..91b4f09 100644 --- a/daemon/src/store/tasks.cpp +++ b/daemon/src/store/tasks.cpp @@ -19,7 +19,7 @@ constexpr const char* kCols = "size_bytes, downloaded_bytes, resumable, " "req_segments, eff_segments, req_buffer_bytes, eff_buffer_bytes, queue_position, " "error_code, error_message, error_http_status, error_retryable, error_attempt, " - "error_next_retry_at"; + "error_next_retry_at, speed_bps"; DbResult bind_opt(Stmt& s, int i, const std::optional& v) { return v ? s.bind(i, std::string_view(*v)) : s.bind_null(i); @@ -72,6 +72,7 @@ TaskRow read_row(Stmt& s) { if (!s.column_is_null(30)) r.error_retryable = s.column_int(30) != 0; r.error_attempt = col_opt_int(s, 31); r.error_next_retry_at = col_opt_text(s, 32); + r.speed_bps = s.column_int(33); return r; } @@ -103,9 +104,9 @@ DbResult Tasks::insert(const TaskRow& r) { "checksum_algo, checksum_value, size_bytes, downloaded_bytes, resumable, " "req_segments, eff_segments, req_buffer_bytes, eff_buffer_bytes, queue_position, " "error_code, error_message, error_http_status, error_retryable, error_attempt, " - "error_next_retry_at) VALUES(" + "error_next_retry_at, speed_bps) VALUES(" "?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22," - "?23,?24,?25,?26,?27,?28,?29,?30,?31,?32,?33)"); + "?23,?24,?25,?26,?27,?28,?29,?30,?31,?32,?33,?34)"); if (!st) return std::unexpected(st.error()); auto chk = [](DbResult r) { return r.has_value(); }; @@ -134,7 +135,8 @@ DbResult Tasks::insert(const TaskRow& r) { chk(r.error_retryable ? st->bind(31, static_cast(*r.error_retryable)) : st->bind_null(31)) && chk(bind_opt(*st, 32, r.error_attempt)) && - chk(bind_opt(*st, 33, r.error_next_retry_at)); + chk(bind_opt(*st, 33, r.error_next_retry_at)) && + chk(st->bind(34, r.speed_bps)); if (!ok) return std::unexpected(DbError{0, "failed to bind a task column"}); if (auto r2 = st->step(); !r2) return std::unexpected(r2.error()); @@ -265,15 +267,45 @@ DbResult Tasks::remove(std::string_view task_id) { } DbResult Tasks::update_progress(std::string_view task_id, std::int64_t downloaded_bytes, - std::int64_t eff_segments, std::int64_t eff_buffer_bytes) { + std::int64_t speed_bps, std::int64_t eff_segments, + std::int64_t eff_buffer_bytes) { auto st = db_.prepare( - "UPDATE tasks SET downloaded_bytes = ?2, eff_segments = ?3, eff_buffer_bytes = ?4 " - "WHERE task_id = ?1"); + "UPDATE tasks SET downloaded_bytes = ?2, speed_bps = ?3, eff_segments = ?4, " + "eff_buffer_bytes = ?5 WHERE task_id = ?1"); if (!st) return std::unexpected(st.error()); if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); if (auto r = st->bind(2, downloaded_bytes); !r) return std::unexpected(r.error()); - if (auto r = st->bind(3, eff_segments); !r) return std::unexpected(r.error()); - if (auto r = st->bind(4, eff_buffer_bytes); !r) return std::unexpected(r.error()); + if (auto r = st->bind(3, speed_bps); !r) return std::unexpected(r.error()); + if (auto r = st->bind(4, eff_segments); !r) return std::unexpected(r.error()); + if (auto r = st->bind(5, eff_buffer_bytes); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + +DbResult Tasks::set_probe_result(std::string_view task_id, const ProbeFields& f) { + auto st = db_.prepare( + "UPDATE tasks SET size_bytes = ?2, resumable = ?3, etag = ?4, last_modified = ?5, " + "content_type = ?6, effective_url = ?7 WHERE task_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = bind_opt(*st, 2, f.size_bytes); !r) return std::unexpected(r.error()); + if (auto r = st->bind(3, static_cast(f.resumable)); !r) + return std::unexpected(r.error()); + if (auto r = bind_opt(*st, 4, f.etag); !r) return std::unexpected(r.error()); + if (auto r = bind_opt(*st, 5, f.last_modified); !r) return std::unexpected(r.error()); + if (auto r = bind_opt(*st, 6, f.content_type); !r) return std::unexpected(r.error()); + if (auto r = bind_opt(*st, 7, f.effective_url); !r) return std::unexpected(r.error()); + if (auto r = st->step(); !r) return std::unexpected(r.error()); + return sqlite3_changes(db_.raw()) > 0; +} + +DbResult Tasks::set_final_bytes(std::string_view task_id, std::int64_t bytes) { + auto st = db_.prepare( + "UPDATE tasks SET downloaded_bytes = ?2, size_bytes = COALESCE(size_bytes, ?2) " + "WHERE task_id = ?1"); + if (!st) return std::unexpected(st.error()); + if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error()); + if (auto r = st->bind(2, bytes); !r) return std::unexpected(r.error()); if (auto r = st->step(); !r) return std::unexpected(r.error()); return sqlite3_changes(db_.raw()) > 0; } @@ -296,9 +328,16 @@ proto::TaskSummary to_summary(const TaskRow& r) { s.sizeBytes = r.size_bytes; s.downloadedBytes = r.downloaded_bytes; if (auto st = proto::parse_TaskState(r.state)) s.state = *st; - s.speedBps = 0; + s.speedBps = r.speed_bps; s.resumable = r.resumable; - s.segments = r.eff_segments; + // TaskSummary.segments is minimum:1, always -- even a task that has never connected + // reports the count it WOULD use (its requested value, or the frozen default), + // never the "not started yet" placeholder of 0 that used to leak onto the wire. + s.segments = r.eff_segments > 0 ? r.eff_segments + : r.req_segments && *r.req_segments > 0 ? *r.req_segments + : 8; + if (s.segments < 1) s.segments = 1; + if (s.segments > 32) s.segments = 32; s.categoryId = r.category_id; s.queueId = r.queue_id; s.queuePosition = r.queue_position; diff --git a/daemon/src/store/tasks.hpp b/daemon/src/store/tasks.hpp index cecf8b5..71b33db 100644 --- a/daemon/src/store/tasks.hpp +++ b/daemon/src/store/tasks.hpp @@ -39,6 +39,7 @@ struct TaskRow { std::optional size_bytes; std::int64_t downloaded_bytes = 0; + std::int64_t speed_bps = 0; bool resumable = false; std::optional req_segments; @@ -81,7 +82,26 @@ public: // Byte-counter update from an engine progress tick — cheaper than a full row rewrite, // and keeps download.list / download.get current between state transitions. DbResult update_progress(std::string_view task_id, std::int64_t downloaded_bytes, - std::int64_t eff_segments, std::int64_t eff_buffer_bytes); + std::int64_t speed_bps, std::int64_t eff_segments, + std::int64_t eff_buffer_bytes); + + // What the probe learned, persisted before start() so a task that completes before any + // progress tick still reports a real sizeBytes / resumable (not the pre-probe default). + struct ProbeFields { + std::optional size_bytes; + bool resumable = false; + std::optional etag; + std::optional last_modified; + std::optional content_type; + std::optional effective_url; + }; + DbResult set_probe_result(std::string_view task_id, const ProbeFields& fields); + + // on_finished's byte count, for a task that completes before any progress tick ever + // ran (see AGENT-DAEMON review: the bug this closes). size_bytes is only filled in if + // still unset — the probe's total_size is the more authoritative source when both + // exist and happen to disagree (a chunked source with no declared length, say). + DbResult set_final_bytes(std::string_view task_id, std::int64_t bytes); private: Db& db_; diff --git a/daemon/tests/store_migrations_test.cpp b/daemon/tests/store_migrations_test.cpp index 50971aa..76e97d9 100644 --- a/daemon/tests/store_migrations_test.cpp +++ b/daemon/tests/store_migrations_test.cpp @@ -100,11 +100,20 @@ void run() { } // --- forward-only: from every released version [0 .. head-1], reach head -------- + // A "released version N" db has the real schema migrations 1..N actually built, not + // just the pragma set to N — faking the pragma alone left `start >= 1` cases running + // a later migration (e.g. 0002's ALTER TABLE tasks / rebuild of segments) against a + // db with no tables at all. for (std::int64_t start = 0; start < head; ++start) { auto db = Db::open(":memory:"); CHECK(db.has_value()); if (!db) continue; - CHECK(db->set_user_version(start).has_value()); + for (const auto& m : embedded_migrations()) { + if (m.version > start) break; + CHECK(db->exec(m.sql).has_value()); + CHECK(db->set_user_version(m.version).has_value()); + } + CHECK_EQ(db->user_version(), start); auto out = migrate_to_head(*db); CHECK(out.has_value()); if (out) {