diff --git a/core/docs/engine-api-m1.md b/core/docs/engine-api-m1.md new file mode 100644 index 0000000..85ebb4a --- /dev/null +++ b/core/docs/engine-api-m1.md @@ -0,0 +1,186 @@ +# CORE → DAEMON — the engine API (M1) + +**Status: reviewed and signed off** by DAEMON on 2026-09-10 +(`daemon/docs/engine-api-review.md`). Nothing forced a `sched/` or dispatch rewrite. The +five open questions are resolved at the bottom; the review's four confirms are folded into +the headers (`sha512` added to `Checksum::Algo`; the parent-directory contract made +explicit; `cancel()` ordering documented; `rate_limiter()` accessor added). + +Integration order: DAEMON wires this in after `daemon/src/sched/` lands (the scheduler is +what calls `start`/`pause`/`resume`/`cancel` and drives `set_task_order`); `sched/` builds +against these headers in parallel with CORE stage 8 and does not need the bodies. + +--- + +This is the `libveloxcore` public surface `veloxd` links and calls +to actually run a download. It is the thing the AGENT-CORE brief asked for on day one and +that slipped: DAEMON has an RPC surface and a store and, until this is agreed, nothing to +call. Sketch headers: `core/include/vdm/engine.hpp`, `core/include/vdm/task/download.hpp` +(both compile now; `Engine` / `DownloadHandle` bodies land in CORE stage 8). Nothing here +touches `contracts/` — DAEMON projects these callbacks onto `TaskSummary` / `TaskDetail` / +`event.*`. + +Please review the field semantics, the threading/lifetime rules, and the +pause/resume/cancel contract, and raise anything that would force a `daemon/src/sched/` or +RPC-dispatch rewrite later. Open questions are at the bottom. + +--- + +## 1. `DownloadSpec` — what DAEMON hands the engine + +DAEMON has already run the rules engine, picked the category folder, canonicalised the +path and checked it against the allowed roots (`-32011` is DAEMON's error, raised before +`start()`), and resolved the filename. The engine's spec is the concrete result. + +| field | who fills it | notes | +|---|---|---| +| `url`, `mirrors` | DAEMON | `mirrors` are alternative URLs for the *same bytes*; the segmenter's requeue prefers a different host after 3 connection failures (docs/04 §3). | +| `headers`, `cookies`, `referrer`, `user_agent` | DAEMON, verbatim from the capture | replayed on every segment request and on the probe. | +| `save_path` | DAEMON | **absolute and final.** The engine never canonicalises or root-checks. `.veloxpart` and `.veloxpart.meta` sit beside it; on success the part file is renamed in place. | +| `segments` | user / `connection.maxSegmentsPerDownload` | requested upper bound 1–32; the engine lowers it to the per-host cap and to 1 for a non-resumable source. Effective value comes back in `Progress.effective_segments`. | +| `buffer_bytes` | user / `connection.bufferBytes` | requested per-segment; silently reduced to fit `maxTotalBufferBytes` across all live segments. Effective value in `Progress.effective_buffer_bytes` → your `TaskDetail.effectiveBufferBytes`. | +| `proxy`, `auth` | DAEMON | `auth` carries credentials only if known up front (Secret Service). Leave `scheme == none` to get an `on_auth_required` on a 401/407 instead. | +| `checksum` | user (`download.add.checksum`) | `{algo, hex}`. Verified during `verifying`; a mismatch is a terminal `failed` with `Error::checksum_mismatch`. | +| `probe_hint` | DAEMON | the `ProbeResult` you already got for the File Info dialog. Supplying it skips the engine's own probe — the task starts in `connecting`, not `probing`. The engine still revalidates with `If-Range` on resume. | +| `allow_resume` | DAEMON | `true`: if a CRC-valid `.veloxpart.meta` sits beside `save_path`, resume from it. `false`: start fresh, overwrite. Your restart flow (ADR 0013 §5) sets this per task. | +| `max_retries` | user / default 10 | per segment, before `failed` with `Error::max_retries_exhausted`. | + +`start()` returns immediately. It never throws and never blocks on the network; a bad URL, +DNS failure, or unwritable `save_path` is delivered through `on_finished`. + +## 2. The state machine the engine drives + +`EngineState` is the CORE-owned subset of the wire `TaskState` (ADR 0013 §1): + +``` +probing ─▶ connecting ⇄ downloading ─▶ assembling ─▶ verifying ─▶ complete + │ │ ▲ │ │ (M4 mux; a no-op rename in M1) + │ │ └─ retry_wait ┘ + ▼ ▼ + (any) ──────▶ paused ──(resume)──▶ connecting + (any CORE state) ──────────────────▶ failed (terminal, engine-initiated) + (any state, on cancel()) ──────────▶ cancelled (terminal, DAEMON-initiated) +``` + +`new` and `queued` are yours; the engine never emits them. `start()` corresponds to your +`queued → probing`. Every transition is reported through `on_state(from, to, error?)`, +including the auto-pauses (§4) and the terminals. `previousState` on your +`event.task.state` maps straight from the `from` argument. + +## 3. Threading and lifetime + +- **Callbacks run on an engine thread** — a transfer worker, the progress timer, or a + dispatch thread — **never** the thread that called `start()` / `pause()` / etc. +- **Per task, callbacks are serialised.** You will never get two callbacks for the same + handle at once. Across tasks they run concurrently. +- **A callback must not block.** It runs on a thread doing real transfer work; a slow + callback stalls that work. Hand off to your own queue/loop. +- **A callback must not re-enter the same handle synchronously** — no `pause()` from + inside `on_state`, etc. Post it. (Calling into a *different* handle, or into + `segment_budget()`, is fine.) +- **`on_finished` is always the last callback** for a task. After it returns the engine + makes no further callbacks for that handle and the handle's control methods are no-ops. +- **The handle is copyable and thread-safe.** Dropping the last copy does **not** cancel — + the task runs on. Call `cancel()` to stop it. (DAEMON holds the handle for the task's + life anyway.) +- **`Engine` must outlive every handle.** `~Engine()` cancels all running tasks and joins + their workers before returning — expect it to block briefly. +- **Logging**: the engine writes through `vdm::set_log_sink()` (a `core/util` global). + Install your sink once at startup; the engine never opens a file itself. + +## 4. `paused` is shared, and idempotency is the contract (ADR 0013 §2) + +Both sides put a task in `paused`, for disjoint reasons: + +- **DAEMON-initiated**: `handle.pause()` — user pause, a schedule window closing, a queue + stop, `Queue.onComplete`, the admission governor reconciling a lowered + `maxActiveSegments`. +- **Engine auto-pause**: `on_auth_required` (401/407), `on_decision_needed` + (`server_file_changed` / stale range), disk full. The engine transitions to `paused` + on its own and fires `on_state(_, paused, ErrorInfo{...})` — the same path as any other + transition. `ErrorInfo.code` present ⇒ engine-initiated; absent ⇒ you did it. That is + the only discriminator, and it is what your `error`-on-`paused` widening (open item 3 of + ADR 0013) carries on the wire. + +**Idempotency, now a signature:** + +| call | already in that state / terminal | otherwise | +|---|---|---| +| `pause()` | no-op, no error | stop new segment requests, flush + `fdatasync` in-flight buffers, write `.veloxpart.meta`, release the budget slots, `on_state(_, paused, nullopt)`. Bounded by the slowest in-flight flush. | +| `resume()` | no-op if not `paused`; no-op if terminal | revalidate with `If-Range`, re-acquire budget slots, `paused → connecting`, resume from the sidecar offsets. | +| `cancel(discard_partial)` | no-op if already terminal | stop everything, `on_state(_, cancelled, nullopt)`, `on_finished(Err{cancelled})`. `discard_partial` also unlinks `.veloxpart[.meta]` — wire this to `download.remove {deleteFile}`. | + +`resume()` after an auto-pause for `auth_required` **without** a preceding +`provide_auth()` is a no-op — the task stays paused. This is ADR 0013 §3's "resume must +not cross reasons", enforced on CORE's side: the scheduler cannot accidentally un-pause a +task waiting on credentials. + +- **`provide_auth(user, pass, remember)`** — acts only while the task is auto-paused for + auth; supplies the credential for the retry and resumes. `remember` asks *you* to + persist to the Secret Service; the engine never stores it. No-op otherwise. +- **`decide(Decision)`** — acts only while auto-paused for a decision. `restart` discards + the partial and re-downloads; `keep_partial` continues against what is on disk (the + user's stated risk); `abort` → `failed`. No-op otherwise. +- **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": the engine re-probes + the new URL to validate it, then **restarts every segment on it** (the signed-URL-expiry + case the wire method exists for), keeping the bytes already on disk. Mirror rotation is + *not* this — that is `spec.mirrors` + the segmenter's requeue-to-a-different-host. +- **`on_decision_needed`** is only for the cases the engine cannot resolve itself. A + routine 416 / stale range is the engine's own re-probe + re-split loop; it escalates + `DecisionRequest{range_metadata_stale}` **only when that loop fails**, at which point + "retry the same range" is already exhausted — so `{restart, keep_partial, abort}` is the + whole choice set. + +## 5. Progress + +`on_progress` is coalesced to **≤ 4 Hz per task** inside the engine — the same cadence as +`event.task.progress`, so your batcher can forward without re-throttling. It carries +aggregate `downloaded` / `speed_bps` / `eta_seconds`, the effective segment count and +buffer size, and a `SegmentProgress[]` (index, inclusive `[start,end]`, `completed`, +per-segment speed, state) for the GUI's segment bars. `total` is absent for a chunked +source until the stream ends. + +`handle.state()` and `handle.progress()` are synchronous lock-guarded snapshots for +`download.get` / `download.list` — call them any time, including from your RPC thread. + +## 6. What the engine does NOT do + +- No filename resolution, no category matching, no path canonicalisation, no allowed-root + check — all DAEMON, before `start()`. +- No queueing, scheduling, priority, or "when queue completes" — DAEMON, via + `segment_budget().set_task_order()` and by choosing when to call `start()` / `pause()`. +- No persistence beyond `.veloxpart.meta`. On a daemon restart the engine knows nothing; + you reload from SQLite, rewrite CORE-owned states to `queued`, and re-`start()` with + `allow_resume = true` (ADR 0013 §5). +- No credential storage. Ever (`CLAUDE.md` §4). + +--- + +## Resolved (DAEMON review, 2026-09-10) + +1. **`probe_hint` stays optional.** DAEMON has a `ProbeResult` only on the File-Info path; + capture-take, `velox add`, `addBatch` and restart have none. The engine probes when + it's absent. +2. **One `cancel(discard_partial)`.** `download.cancel` = `cancel(false)`; + `download.remove` = `cancel(true)` for a live task (plus DAEMON's row/file cleanup), or + pure DAEMON-side for an already-terminal one. No `handle.remove()`. +3. **`{restart, keep_partial, abort}` is the whole set.** The engine owns the routine + 416 re-probe/re-split and only escalates `on_decision_needed` when that loop fails — + "retry the same range" is exhausted by then. +4. **Per-task 4 Hz is fine.** DAEMON coalesces across tasks for `event.task.progress` + regardless; `on_progress_batch` is a nice-to-have and must not block stage 8. +5. **`refresh_url` restarts all segments on the new URL** after a validating re-probe (the + signed-URL case). Mirror rotation is `spec.mirrors` + the segmenter, not this. + +## Review confirms, folded in + +- **(a)** `vdm::TaskId` is a cheap-copy hashable value; DAEMON never constructs one — it + only receives it from `start()` / callbacks and passes it back to `set_task_order()` + etc. ✔ (`vdm/ids.hpp`) +- **(b)** DAEMON `mkdir -p`s `save_path`'s parent before `start()`. The engine opens the + file and fails with `Error::path_rejected` if the directory is missing. ✔ (documented on + `DownloadSpec`) +- **(c)** `Checksum::Algo` now has `sha512`, matching the wire `Checksum` set. ✔ +- **(d)** `cancel()` always fires `on_state(_, cancelled, nullopt)` then + `on_finished(Err{Error::canceled})` (note: the taxonomy value is `canceled`), in that + order. ✔ (documented on `DownloadHandle::cancel`) diff --git a/core/include/vdm/README.md b/core/include/vdm/README.md index 7bbdfda..18276dd 100644 --- a/core/include/vdm/README.md +++ b/core/include/vdm/README.md @@ -1,8 +1,12 @@ # `libveloxcore` — public API -**Status: M1 in progress.** Only `util/` is landed. The download-facing API -(`DownloadSpec`, `DownloadTask`, probe, typed callbacks) arrives with later stages and -is reviewed by DAEMON before M2 (AGENT-CORE DoD). +**Status: M1 in progress.** `util/`, `net/` (http_client, probe, url, content_disposition), +`io/` (sparse_file, write_buffer), `meta/veloxpart`, and `segment/` (segmenter, budget) +are landed. The **download entry point** — `vdm::Engine`, `vdm::task::DownloadSpec` / +`DownloadHandle` / `DownloadCallbacks` — is sketched in `vdm/engine.hpp` and +`vdm/task/download.hpp` and **out for DAEMON review**: see +[`core/docs/engine-api-m1.md`](../../docs/engine-api-m1.md). Bodies land in CORE stage 8; +build against the value types now. Layering (CLAUDE.md §3): this library knows nothing about JSON, SQL, Qt, or RPC. Input is a spec value; output is bytes on disk plus typed callbacks. DAEMON projects engine state diff --git a/core/include/vdm/engine.hpp b/core/include/vdm/engine.hpp new file mode 100644 index 0000000..fdc3628 --- /dev/null +++ b/core/include/vdm/engine.hpp @@ -0,0 +1,82 @@ +// vdm/engine.hpp — the download engine's single entry point. REVIEW SKETCH (stage 7 +// pre-work); bodies land in stage 8. See core/docs/engine-api-m1.md. +// +// The engine owns the HTTP client, the probe pool, the segment budget, and the disk I/O. +// Its input is a DownloadSpec; its output is bytes at save_path plus typed callbacks. No +// JSON, no SQL, no Qt, no RPC — DAEMON projects the callbacks onto the wire contract. +// +// This header compiles standalone. + +#ifndef VDM_ENGINE_HPP +#define VDM_ENGINE_HPP + +#include +#include + +#include "vdm/rate/token_bucket.hpp" +#include "vdm/segment/budget.hpp" +#include "vdm/task/download.hpp" + +namespace vdm { + +class Engine { + public: + struct Config { + // Defaults used when a DownloadSpec leaves the field unset. Live-adjustable via + // the setters below (they take effect on the next segment (re)assignment, not by + // resizing an in-flight buffer). + std::uint32_t default_segments = 8; // connection.maxSegmentsPerDownload + std::uint64_t default_buffer_bytes = 1u << 20; // connection.bufferBytes (1 MiB) + std::uint64_t min_segment_bytes = 1u << 20; // never split below this + std::uint64_t max_total_buffer_bytes = 128ull << 20; // connection.maxTotalBufferBytes + std::uint32_t max_active_segments = 32; // connection.maxActiveSegments + std::uint32_t probe_pool_size = 4; // ADR 0011 §5, outside the budget + long default_max_retries = 10; // per segment + std::uint32_t http_workers = 0; // 0 => hardware-derived (<=4) + }; + + Engine(); // default Config + explicit Engine(Config cfg); + ~Engine(); // cancels every running task and joins before returning + + Engine(const Engine &) = delete; + Engine &operator=(const Engine &) = delete; + + // Start a download. Returns immediately with a handle; the task begins in `probing` + // (or `connecting` when spec.probe_hint is supplied). Every failure — bad URL, DNS, + // an unwritable save_path — is delivered through callbacks.on_finished, never thrown. + [[nodiscard]] task::DownloadHandle start(task::DownloadSpec spec, + task::DownloadCallbacks callbacks); + + // The global segment allocator. DAEMON's scheduler drives admission through this + // (set_max_active_segments / set_host_segment_cap / set_task_order) and reads + // occupancy from it (budget() / segments_active() / starved_tasks() / + // on_budget_changed). See ADR 0011. + [[nodiscard]] segment::SegmentBudget &segment_budget() noexcept; + + // The hierarchical speed limiter (docs/04 §6): global -> per-queue -> per-task token + // buckets. `limiter.set {globalBps, enabled}` -> rate_limiter().set_global_limit(); + // per-queue / per-task limits and the task<->queue attachment come from DAEMON too. + // The engine paces every segment read through it. + [[nodiscard]] rate::RateLimiter &rate_limiter() noexcept; + + // Live settings (connection.* changes from settings.set). Each affects future work. + void set_default_segments(std::uint32_t n); + void set_default_buffer_bytes(std::uint64_t bytes); + void set_max_total_buffer_bytes(std::uint64_t bytes); + void set_probe_pool_size(std::uint32_t n); + + // A standalone probe for the File Info dialog, on the same pool as spec-less probes + // (never charged against the segment budget). capture.offer's 750 ms deadline is + // DAEMON's to enforce — it should answer `ignore` and probe after, never block on + // this. + void probe(net::ProbeRequest req, std::function)> done); + + private: + struct Impl; + std::unique_ptr impl_; +}; + +} // namespace vdm + +#endif // VDM_ENGINE_HPP diff --git a/core/include/vdm/ids.hpp b/core/include/vdm/ids.hpp index 914b4a4..9304ba8 100644 --- a/core/include/vdm/ids.hpp +++ b/core/include/vdm/ids.hpp @@ -24,6 +24,15 @@ struct TaskId { friend constexpr auto operator<=>(const TaskId &, const TaskId &) = default; }; +// A scheduler queue. A task may belong to one — for the per-queue rate limit and +// per-queue concurrency. A task with no queue rate-limits against the global bucket only. +struct QueueId { + std::uint64_t value = 0; + + [[nodiscard]] constexpr bool valid() const noexcept { return value != 0; } + friend constexpr auto operator<=>(const QueueId &, const QueueId &) = default; +}; + using SteadyTime = std::chrono::steady_clock::time_point; } // namespace vdm @@ -35,4 +44,11 @@ struct std::hash { } }; +template <> +struct std::hash { + std::size_t operator()(vdm::QueueId id) const noexcept { + return std::hash{}(id.value); + } +}; + #endif // VDM_IDS_HPP diff --git a/core/include/vdm/rate/token_bucket.hpp b/core/include/vdm/rate/token_bucket.hpp new file mode 100644 index 0000000..49c7100 --- /dev/null +++ b/core/include/vdm/rate/token_bucket.hpp @@ -0,0 +1,186 @@ +// vdm/rate/token_bucket.hpp — a lazily-refilled token bucket, and the global -> queue -> +// task limiter hierarchy built on it (docs/04 §6). +// +// A segment worker calls RateLimiter::acquire(task, n) after receiving n body bytes. If +// every applicable level (task, its queue, global) has n tokens, it consumes n from each +// and returns 0. Otherwise it consumes nothing and returns how long to wait before +// retrying — the worker returns CURL_WRITEFUNC_PAUSE and schedules a curl_easy_pause +// resume after that delay (the "precision" layer on top of CURLOPT_MAX_RECV_SPEED_LARGE). +// +// This header compiles standalone. + +#ifndef VDM_RATE_TOKEN_BUCKET_HPP +#define VDM_RATE_TOKEN_BUCKET_HPP + +#include +#include +#include +#include +#include +#include + +#include "vdm/ids.hpp" + +namespace vdm::rate { + +// rate_bps == 0 means unlimited: consume() always succeeds and never waits. +class TokenBucket { + public: + TokenBucket() = default; + // `burst` caps how many tokens accumulate while idle; 0 => 1 second's worth. + explicit TokenBucket(std::uint64_t rate_bps, std::uint64_t burst = 0) { + set_rate(rate_bps, burst); + } + + void set_rate(std::uint64_t rate_bps, std::uint64_t burst = 0) { + std::lock_guard lk(mu_); + const bool was_unlimited = rate_ == 0; + rate_ = rate_bps; + cap_ = burst ? burst : rate_bps; // 1 s of burst by default + // A freshly-limited bucket starts full: you may transfer at burst speed + // immediately, then it throttles (classic token bucket / IDM behaviour). Lowering + // an existing limit only clamps down — it never hands out a fresh burst. + if (rate_bps > 0 && was_unlimited) + tokens_ = cap_; + else if (tokens_ > cap_) + tokens_ = cap_; + last_ = clock::now(); + } + [[nodiscard]] std::uint64_t rate() const { + std::lock_guard lk(mu_); + return rate_; + } + + // Consume `n` if available; otherwise consume nothing. Returns the wait until `n` + // tokens *would* be available (0 when it consumed). + [[nodiscard]] std::chrono::nanoseconds consume(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return {}; + refill_locked(); + if (tokens_ >= n) { + tokens_ -= n; + return {}; + } + const std::uint64_t deficit = n - tokens_; + // ns to earn `deficit` tokens at rate_ bytes/s + return std::chrono::nanoseconds{ + static_cast((deficit * 1'000'000'000ull + rate_ - 1) / rate_)}; + } + + // Two-phase for the hierarchy: check every level, then commit on all or none. + [[nodiscard]] std::chrono::nanoseconds peek(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return {}; + refill_locked(); + if (tokens_ >= n) + return {}; + const std::uint64_t deficit = n - tokens_; + return std::chrono::nanoseconds{ + static_cast((deficit * 1'000'000'000ull + rate_ - 1) / rate_)}; + } + void commit(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return; + tokens_ = tokens_ >= n ? tokens_ - n : 0; + } + + private: + using clock = std::chrono::steady_clock; + void refill_locked() { + auto now = clock::now(); + auto dt = std::chrono::duration_cast(now - last_).count(); + if (dt <= 0) + return; + last_ = now; + // added = rate_ * dt / 1e9, guarding overflow for very long idle gaps + long double added = + static_cast(rate_) * static_cast(dt) / 1'000'000'000.0L; + std::uint64_t add = + added >= static_cast(cap_) ? cap_ : static_cast(added); + tokens_ = tokens_ + add > cap_ ? cap_ : tokens_ + add; + } + + mutable std::mutex mu_; + std::uint64_t rate_ = 0; + std::uint64_t cap_ = 0; + std::uint64_t tokens_ = 0; + clock::time_point last_ = clock::now(); +}; + +// The hierarchy. All limits default to 0 (unlimited). A task with no queue is limited by +// task + global only. +class RateLimiter { + public: + void set_global_limit(std::uint64_t bps) { global_.set_rate(bps); } + [[nodiscard]] std::uint64_t global_limit() const { return global_.rate(); } + + void set_queue_limit(QueueId q, std::uint64_t bps) { + std::lock_guard lk(mu_); + queues_[q].set_rate(bps); + } + void set_task_limit(TaskId t, std::uint64_t bps) { + std::lock_guard lk(mu_); + tasks_[t].set_rate(bps); + } + + void attach_task(TaskId t, std::optional q) { + std::lock_guard lk(mu_); + tasks_.try_emplace(t); + if (q) { + task_queue_[t] = *q; + queues_.try_emplace(*q); + } else { + task_queue_.erase(t); + } + } + void detach_task(TaskId t) { + std::lock_guard lk(mu_); + tasks_.erase(t); + task_queue_.erase(t); + } + + // Consume `n` bytes against task, queue and global. 0 => consumed everywhere. > 0 => + // consumed nowhere; wait that long and retry. Held under mu_ for its whole duration + // so a concurrent detach_task() can't invalidate the bucket it is using. + [[nodiscard]] std::chrono::nanoseconds acquire(TaskId t, std::uint64_t n) { + std::lock_guard lk(mu_); + TokenBucket *tb = nullptr; + TokenBucket *qb = nullptr; + if (auto it = tasks_.find(t); it != tasks_.end()) + tb = &it->second; + if (auto qit = task_queue_.find(t); qit != task_queue_.end()) + if (auto q = queues_.find(qit->second); q != queues_.end()) + qb = &q->second; + + // peek all, then commit all or none — no level "leaks" tokens on a partial miss. + std::chrono::nanoseconds wait{}; + if (tb) + wait = std::max(wait, tb->peek(n)); + if (qb) + wait = std::max(wait, qb->peek(n)); + wait = std::max(wait, global_.peek(n)); + if (wait.count() > 0) + return wait; + + if (tb) + tb->commit(n); + if (qb) + qb->commit(n); + global_.commit(n); + return {}; + } + + private: + mutable std::mutex mu_; // guards the maps AND serialises acquire() + TokenBucket global_; + std::unordered_map queues_; + std::unordered_map tasks_; + std::unordered_map task_queue_; +}; + +} // namespace vdm::rate + +#endif // VDM_RATE_TOKEN_BUCKET_HPP diff --git a/core/include/vdm/task/download.hpp b/core/include/vdm/task/download.hpp new file mode 100644 index 0000000..4eb184b --- /dev/null +++ b/core/include/vdm/task/download.hpp @@ -0,0 +1,218 @@ +// vdm/task/download.hpp — the public download API: what DAEMON hands the engine and how +// the engine reports back. REVIEW SKETCH (stage 7 pre-work) — value types are final +// enough to build against; Engine/DownloadHandle bodies land in stage 8. +// +// See core/docs/engine-api-m1.md for the threading, lifetime, and pause/resume/cancel +// contract that goes with these signatures. +// +// This header compiles standalone. + +#ifndef VDM_TASK_DOWNLOAD_HPP +#define VDM_TASK_DOWNLOAD_HPP + +#include +#include +#include +#include +#include +#include +#include + +#include "vdm/ids.hpp" +#include "vdm/net/http_types.hpp" +#include "vdm/net/probe.hpp" +#include "vdm/segment/segmenter.hpp" +#include "vdm/util/error.hpp" +#include "vdm/util/result.hpp" + +namespace vdm { +class Engine; // owns and fills DownloadHandle (see vdm/engine.hpp) +} // namespace vdm + +namespace vdm::task { + +// --- input --------------------------------------------------------------------------- + +struct Checksum { + enum class Algo { md5, sha1, sha256, sha512 }; // matches the wire Checksum set + Algo algo = Algo::sha256; + std::string hex; // lower-case, no separators +}; + +// Everything the engine needs to run ONE download. DAEMON has already run the rules +// engine, canonicalised the path, checked it against the allowed roots, resolved the +// filename, and created the parent directory — `save_path` is absolute and final and its +// directory exists. `.veloxpart` and `.veloxpart.meta` live beside +// it during the transfer; on success the part file is renamed in place. If the directory +// is missing at open time the task fails with Error::path_rejected. +struct DownloadSpec { + std::string url; + std::vector mirrors; // alternative URLs for the same bytes + + std::vector headers; // the browser's, verbatim + std::vector cookies; + std::string referrer; + std::string user_agent; + + std::string save_path; // absolute; the engine never canonicalises or root-checks + + std::optional segments; // requested 1..32; nullopt => engine default + std::optional buffer_bytes; // requested per segment; nullopt => default + + net::ProxyConfig proxy; + net::AuthConfig auth; // credentials known up front (e.g. from the Secret Service); + // leave scheme == none to be prompted on a 401/407 + + std::optional checksum; // verified during `verifying`; mismatch => failed + + // DAEMON usually probed already for the File Info dialog. Pass it to skip a second + // probe; the engine still revalidates on resume. nullopt => the engine probes. + std::optional probe_hint; + + bool allow_resume = true; // if a valid .veloxpart.meta sits beside save_path, resume + // from it; false starts fresh and overwrites + std::optional max_retries; // per-segment; nullopt => engine default (10) +}; + +// --- lifecycle (the CORE-owned subset of the wire TaskState; ADR 0013 §1) ------------- + +enum class EngineState { + probing, + connecting, + downloading, + paused, // shared with DAEMON; entered by either side, idempotently + retry_wait, // the engine's own backoff timer + assembling, // no-op rename in M1; a real mux step for HLS/DASH (M4) + verifying, // checksum + complete, // terminal + failed, // terminal + cancelled, // terminal; always DAEMON- or user-initiated +}; + +[[nodiscard]] constexpr bool is_terminal(EngineState s) noexcept { + return s == EngineState::complete || s == EngineState::failed || s == EngineState::cancelled; +} + +// --- progress --------------------------------------------------------------------------- + +struct SegmentProgress { + std::uint32_t index = 0; + std::uint64_t start = 0; + std::uint64_t end = 0; // inclusive + std::uint64_t completed = 0; + std::uint64_t speed_bps = 0; + segment::SegState state = segment::SegState::idle; +}; + +struct Progress { + std::uint64_t downloaded = 0; + std::optional total; // absent for a chunked source until it ends + std::uint64_t speed_bps = 0; // aggregate over the last window + std::optional eta_seconds; + + std::uint32_t effective_segments = 0; // slots the budget granted (held) + std::uint64_t effective_buffer_bytes = 0; // per segment, after the maxTotal clamp + std::vector segments; +}; + +// --- interaction callbacks ---------------------------------------------------------- + +// A 401/407. The task has already auto-paused (state -> paused, error == auth_required). +// DAEMON collects credentials and calls handle.provide_auth(). +struct AuthChallenge { + std::string host; + std::string realm; + enum class Scheme { basic, digest, ntlm, negotiate, unknown }; + Scheme scheme = Scheme::unknown; +}; + +// The server's copy changed under us (a 200 where a 206 was expected, or an If-Range / +// ETag mismatch on resume — docs/04 §5), or the range metadata went stale (416). The +// task has auto-paused. DAEMON asks the user and calls handle.decide(). +struct DecisionRequest { + enum class Kind { server_file_changed, range_metadata_stale }; + Kind kind = Kind::server_file_changed; + std::string detail; // human-readable, for the dialog body +}; + +enum class Decision { + restart, // discard the partial file, download again from scratch + keep_partial, // trust what is on disk and continue (the user's risk) + abort, // give up: the task goes to `failed` +}; + +struct DownloadOutcome { + std::string final_path; + std::uint64_t bytes = 0; + std::optional sha256_hex; // present when a checksum was requested/derived + std::chrono::milliseconds elapsed{0}; +}; + +// All callbacks are optional. See core/docs/engine-api-m1.md for the rules; in short: +// they arrive on an engine thread, are serialised per task, must not block, and must not +// re-enter THIS task's handle synchronously. +struct DownloadCallbacks { + // Coalesced to <= 4 Hz per task (matches the wire event.task.progress cadence). + std::function on_progress; + + // Every lifecycle transition, including the auto-pauses above (to == paused with a + // populated ErrorInfo) and terminals. + std::function &)> + on_state; + + std::function on_auth_required; + std::function on_decision_needed; + + // Fired exactly once, last. Success carries the outcome; failure carries the mapped + // ErrorInfo. After it returns the engine makes no further callbacks for this task and + // the handle's control methods become no-ops. + std::function)> on_finished; +}; + +// --- the handle ------------------------------------------------------------------------- + +// Copyable (shared state). Every method is safe to call from any thread; each posts to +// the engine and returns immediately. Dropping the last handle does NOT cancel the task — +// call cancel() for that. Bodies land in stage 8. +class DownloadHandle { + public: + DownloadHandle() = default; + + [[nodiscard]] TaskId id() const noexcept; + [[nodiscard]] bool valid() const noexcept { return static_cast(state_); } + + // Idempotent. pause() on an already-paused or terminal task is a no-op (no error); + // likewise resume() on a task that is not paused. The resulting state is observed via + // on_state / this->state(), never a return value (ADR 0013 §2). + void pause(); + void resume(); + // Idempotent, terminal. discard_partial also removes the .veloxpart[.meta] files. + // Always fires on_state(_, cancelled, nullopt) then on_finished(Err{Error::canceled}), + // in that order. `download.cancel` == cancel(false); `download.remove` == cancel(true) + // (plus DAEMON's own row/file cleanup). + void cancel(bool discard_partial = false); + + // Only act while the task is awaiting the matching input (auto-paused for auth / + // decision); otherwise a no-op. `remember` asks DAEMON to persist to the Secret + // Service — the engine never stores a credential. + void provide_auth(std::string username, std::string password, bool remember); + void decide(Decision d); + + // IDM's "Refresh Download Address": swap the URL (e.g. a fresh signed URL) on a live + // or paused task without losing progress. Empty `headers` keeps the current ones. + void refresh_url(std::string url, std::vector headers = {}); + + // Synchronous snapshots — cheap, lock-guarded, safe any time. + [[nodiscard]] EngineState state() const; + [[nodiscard]] Progress progress() const; + + private: + friend class vdm::Engine; + struct State; + explicit DownloadHandle(std::shared_ptr s) : state_(std::move(s)) {} + std::shared_ptr state_; +}; + +} // namespace vdm::task + +#endif // VDM_TASK_DOWNLOAD_HPP diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 28db7d1..ca11711 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -30,6 +30,8 @@ vdm_add_test(veloxcore_write_buffer_test io/write_buffer_test.cpp) vdm_add_test(veloxcore_veloxpart_test meta/veloxpart_test.cpp) vdm_add_test(veloxcore_segmenter_test segment/segmenter_test.cpp) vdm_add_test(veloxcore_budget_test segment/budget_test.cpp) +vdm_add_test(veloxcore_engine_api_test task/api_compiles_test.cpp) +vdm_add_test(veloxcore_token_bucket_test rate/token_bucket_test.cpp) set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py) foreach(net_it http_client probe) diff --git a/core/tests/rate/token_bucket_test.cpp b/core/tests/rate/token_bucket_test.cpp new file mode 100644 index 0000000..a0cf443 --- /dev/null +++ b/core/tests/rate/token_bucket_test.cpp @@ -0,0 +1,145 @@ +#include "vdm/rate/token_bucket.hpp" + +#include +#include +#include +#include + +#include "vtest.hpp" + +using namespace vdm; +using namespace vdm::rate; +using namespace std::chrono_literals; + +namespace { +TaskId tid(std::uint64_t v) { + return TaskId{v}; +} +QueueId qid(std::uint64_t v) { + return QueueId{v}; +} +} // namespace + +VT_TEST(tb_unlimited_never_waits) { + TokenBucket b(0); + for (int i = 0; i < 1000; ++i) + VT_CHECK_EQ(b.consume(1'000'000).count(), 0); +} + +VT_TEST(tb_burst_then_throttle) { + // 1000 B/s, default burst = 1 s = 1000 tokens. + TokenBucket b(1000); + VT_CHECK_EQ(b.consume(1000).count(), 0); // drains the burst + auto w = b.consume(1000); // empty now: must wait ~1 s + VT_CHECK(w >= 900ms && w <= 1100ms); +} + +VT_TEST(tb_refills_over_time) { + TokenBucket b(10'000, /*burst=*/10'000); + VT_CHECK_EQ(b.consume(10'000).count(), 0); + std::this_thread::sleep_for(120ms); // ~1200 tokens back + auto w = b.consume(1000); + VT_CHECK_EQ(w.count(), 0); // affordable from the refill + auto w2 = b.consume(5000); + VT_CHECK(w2.count() > 0); // not that much yet +} + +VT_TEST(tb_burst_caps_accumulation) { + TokenBucket b(1000, /*burst=*/2000); + std::this_thread::sleep_for(100ms); // idle far longer than burst/rate would fill + std::this_thread::sleep_for(100ms); + VT_CHECK_EQ(b.consume(2000).count(), 0); // at most the 2000 cap accumulated + VT_CHECK(b.consume(1).count() > 0); // and no more +} + +VT_TEST(tb_set_rate_zero_makes_unlimited) { + TokenBucket b(1000); + VT_CHECK_EQ(b.consume(1000).count(), 0); + VT_CHECK(b.consume(1000).count() > 0); + b.set_rate(0); + VT_CHECK_EQ(b.consume(1'000'000).count(), 0); +} + +// --- the hierarchy -------------------------------------------------------------------- + +VT_TEST(rl_all_unlimited_by_default) { + RateLimiter rl; + rl.attach_task(tid(1), std::nullopt); + for (int i = 0; i < 100; ++i) + VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0); +} + +VT_TEST(rl_tightest_level_binds) { + RateLimiter rl; + rl.set_global_limit(100'000); + rl.set_queue_limit(qid(9), 20'000); + rl.set_task_limit(tid(1), 50'000); + rl.attach_task(tid(1), qid(9)); + + // burst: task 50k, queue 20k, global 100k -> the queue's 20k is the ceiling + VT_CHECK_EQ(rl.acquire(tid(1), 20'000).count(), 0); + auto w = rl.acquire(tid(1), 5'000); + VT_CHECK(w.count() > 0); // queue bucket is dry even though task & global aren't +} + +VT_TEST(rl_no_partial_consumption_on_miss) { + RateLimiter rl; + rl.set_global_limit(1'000'000); // plenty + rl.set_task_limit(tid(1), 1000); // 1 s burst + rl.attach_task(tid(1), std::nullopt); + + VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0); // drain the task bucket + for (int i = 0; i < 5; ++i) + VT_CHECK(rl.acquire(tid(1), 1000).count() > 0); // task bucket blocks, repeatedly + + // global must NOT have been charged for any of those blocked attempts: a fresh task + // limited only by the global bucket can still spend nearly its whole burst (only the + // one *successful* 1000-byte acquire above was charged). + rl.attach_task(tid(2), std::nullopt); + VT_CHECK_EQ(rl.acquire(tid(2), 990'000).count(), 0); +} + +VT_TEST(rl_detach_then_acquire_is_safe_and_unlimited) { + RateLimiter rl; + rl.set_task_limit(tid(1), 1000); + rl.attach_task(tid(1), std::nullopt); + VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0); + rl.detach_task(tid(1)); + // unknown task -> no task/queue bucket, only global (unlimited here) + VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0); +} + +VT_TEST(rl_enforces_aggregate_rate_under_load) { + RateLimiter rl; + const std::uint64_t rate = 4'000'000; // 4 MB/s global + rl.set_global_limit(rate); + for (std::uint64_t i = 1; i <= 8; ++i) + rl.attach_task(tid(i), std::nullopt); + + std::atomic moved{0}; + auto t0 = std::chrono::steady_clock::now(); + std::vector ws; + for (std::uint64_t i = 1; i <= 8; ++i) { + ws.emplace_back([&, id = tid(i)] { + for (int k = 0; k < 400; ++k) { + std::uint64_t chunk = 16 * 1024; + for (;;) { + auto w = rl.acquire(id, chunk); + if (w.count() == 0) + break; + std::this_thread::sleep_for( + std::min(w, std::chrono::milliseconds(20))); + } + moved.fetch_add(chunk); + } + }); + } + ws.clear(); // join + auto secs = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + + double effective = moved.load() / secs; + // Allow one burst's worth of slop plus scheduling noise: effective rate should sit + // within ~2x of the configured limit, never wildly above. + VT_CHECK(effective <= rate * 2.5); + VT_CHECK(moved.load() == 8u * 400u * 16u * 1024u); +} diff --git a/core/tests/task/api_compiles_test.cpp b/core/tests/task/api_compiles_test.cpp new file mode 100644 index 0000000..97ca398 --- /dev/null +++ b/core/tests/task/api_compiles_test.cpp @@ -0,0 +1,84 @@ +// The engine API sketch must compile and its value types must behave. Engine / +// DownloadHandle bodies land in stage 8; this only exercises the data shapes DAEMON +// builds against. + +#include "vdm/engine.hpp" +#include "vdm/task/download.hpp" + +#include + +#include "vtest.hpp" + +using namespace vdm; +using namespace vdm::task; + +VT_TEST(api_download_spec_defaults) { + DownloadSpec s; + s.url = "https://example.com/big.iso"; + s.save_path = "/home/u/Downloads/big.iso"; + VT_CHECK(s.mirrors.empty()); + VT_CHECK(!s.segments.has_value()); + VT_CHECK(!s.buffer_bytes.has_value()); + VT_CHECK(!s.checksum.has_value()); + VT_CHECK(!s.probe_hint.has_value()); + VT_CHECK(s.allow_resume); + VT_CHECK(s.proxy.kind == net::ProxyKind::none); + VT_CHECK(s.auth.scheme == net::AuthScheme::none); +} + +VT_TEST(api_state_helpers) { + VT_CHECK(is_terminal(EngineState::complete)); + VT_CHECK(is_terminal(EngineState::failed)); + VT_CHECK(is_terminal(EngineState::cancelled)); + VT_CHECK(!is_terminal(EngineState::paused)); + VT_CHECK(!is_terminal(EngineState::downloading)); +} + +VT_TEST(api_callbacks_are_all_optional) { + DownloadCallbacks cb; // every std::function default-constructs empty + VT_CHECK(!cb.on_progress); + VT_CHECK(!cb.on_state); + VT_CHECK(!cb.on_auth_required); + VT_CHECK(!cb.on_decision_needed); + VT_CHECK(!cb.on_finished); + + cb.on_state = [](EngineState, EngineState, const std::optional &) {}; + cb.on_finished = [](Result) {}; + VT_CHECK(cb.on_state && cb.on_finished); +} + +VT_TEST(api_value_types_roundtrip) { + Progress p; + p.downloaded = 1234; + p.total = 5000; + p.effective_segments = 4; + SegmentProgress sp; + sp.index = 0; + sp.end = 1249; + sp.completed = 1234; + p.segments.push_back(sp); + VT_CHECK_EQ(p.segments.size(), 1u); + VT_CHECK_EQ(p.segments[0].end, 1249u); + + DownloadOutcome o; + o.final_path = "/x"; + o.bytes = 5000; + VT_CHECK_EQ(o.bytes, 5000u); + + AuthChallenge a; + a.host = "h"; + a.scheme = AuthChallenge::Scheme::digest; + VT_CHECK(a.scheme == AuthChallenge::Scheme::digest); + + DecisionRequest d; + d.kind = DecisionRequest::Kind::server_file_changed; + d.detail = "changed"; + VT_CHECK(d.kind == DecisionRequest::Kind::server_file_changed); +} + +VT_TEST(api_handle_and_engine_are_move_only_shaped) { + static_assert(!std::is_copy_constructible_v, "Engine is non-copyable"); + static_assert(std::is_copy_constructible_v, "handle is a shared handle"); + DownloadHandle h; // default handle is invalid until Engine::start() fills it + VT_CHECK(!h.valid()); +}