From 479f8823241c5894239aa96e361ebae87c13e9db Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 19:50:28 +0400 Subject: [PATCH 1/4] core: O_NOFOLLOW the download target open DAEMON's safepath-adversarial.md accepts a TOCTOU residual between its canonicalise-and-check and the download starting, on the stated grounds that CORE's O_NOFOLLOW open of the final file closes it. That flag was never actually set: SparseFile::open used O_WRONLY|O_CREAT|O_CLOEXEC, so a symlink swapped in as the final path component after DAEMON's check would be followed and redirect our pwrites outside the allowed roots. Add O_NOFOLLOW. A symlinked leaf now fails the open with ELOOP, which errno_to_error already maps to Error::path_rejected. Regular files and the O_CREAT of a fresh part file are unaffected; resume (existing regular part file) is unaffected. Test that a symlinked destination is rejected rather than silently followed, and that the link target is never touched. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/src/io/sparse_file.cpp | 7 ++++++- core/tests/io/sparse_file_test.cpp | 17 +++++++++++++++++ docs/04-engine-design.md | 10 ++++++++-- 3 files changed, 31 insertions(+), 3 deletions(-) diff --git a/core/src/io/sparse_file.cpp b/core/src/io/sparse_file.cpp index 45e2175..fb98210 100644 --- a/core/src/io/sparse_file.cpp +++ b/core/src/io/sparse_file.cpp @@ -74,7 +74,12 @@ Result SparseFile::open(std::string_view path, const OpenOptions &opts) { return ErrorInfo(Error::internal, "SparseFile already open"); std::string p(path); - int flags = O_WRONLY | O_CREAT | O_CLOEXEC; + // O_NOFOLLOW: the final component of a download target must never be a symlink, on + // create or on resume. DAEMON canonicalises the path and checks it against the allowed + // roots before start(), but a symlink swapped in afterwards would redirect our writes + // outside those roots (daemon/docs/safepath-adversarial.md leans on this open closing + // that TOCTOU window). A symlinked leaf fails here with ELOOP -> Error::path_rejected. + int flags = O_WRONLY | O_CREAT | O_CLOEXEC | O_NOFOLLOW; if (opts.truncate_existing) flags |= O_TRUNC; diff --git a/core/tests/io/sparse_file_test.cpp b/core/tests/io/sparse_file_test.cpp index 1a4b469..ba385d1 100644 --- a/core/tests/io/sparse_file_test.cpp +++ b/core/tests/io/sparse_file_test.cpp @@ -117,6 +117,23 @@ VT_TEST(sparse_open_bad_path_is_path_rejected) { VT_CHECK(!f.is_open()); } +VT_TEST(sparse_symlinked_target_is_rejected) { + // A symlink swapped in as the final path component after DAEMON's canonicalise-and-check + // must not be followed: the open is O_NOFOLLOW, so it fails with ELOOP -> path_rejected + // rather than redirecting our writes through the link. + TempPath link; // the download target the caller hands us + TempPath target; // where the symlink points (would-be victim, outside allowed roots) + VT_REQUIRE(::symlink(target.path.c_str(), link.path.c_str()) == 0); + + SparseFile f; + auto r = f.open(link.path, {.total_size = 4096}); + VT_REQUIRE(!r.has_value()); + VT_CHECK_EQ(r.error().code, Error::path_rejected); + VT_CHECK(!f.is_open()); + // the link target was never created/written through + VT_CHECK_EQ(::access(target.path.c_str(), F_OK), -1); +} + VT_TEST(sparse_ops_on_closed_file_error) { SparseFile f; VT_CHECK_EQ(f.write_at(0, bytes("x")).error().code, Error::internal); diff --git a/docs/04-engine-design.md b/docs/04-engine-design.md index 7626cfe..b2ab3dc 100644 --- a/docs/04-engine-design.md +++ b/docs/04-engine-design.md @@ -61,8 +61,14 @@ Rules: ## 4. Disk I/O — the buffer setting you asked for -One file, opened once, `O_WRONLY`. Each segment `pwrite()`s at its own absolute offset, so -**there is no reassembly pass and no second write of the whole file.** +One file, opened once, `O_WRONLY | O_NOFOLLOW`. Each segment `pwrite()`s at its own absolute +offset, so **there is no reassembly pass and no second write of the whole file.** + +`O_NOFOLLOW` on the part-file open: DAEMON canonicalises the save path and checks it against +the allowed roots before `start()`, but the final component could be swapped for a symlink +in the window between that check and our open. A symlinked leaf is rejected here (`ELOOP` → +`Error::path_rejected`), not followed — it closes the TOCTOU residual that +`daemon/docs/safepath-adversarial.md` accepts on those grounds. - `posix_fallocate()` the full size up front → contiguous extents, no ENOSPC surprise at 99 %, no fragmentation. From afaded85f85704cbb7d0854b3cba2bc8627f661b Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 20:19:05 +0400 Subject: [PATCH 2/4] core: carry credentials through the probe and its auth handshake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit libcurl with CURLAUTH_ANY answers a 401/407 by resending the request with an Authorization header. Two spots in net/ cut that short: - http_client's header callback delivered the response head exactly once and latched `head_delivered`, so after an auth challenge the caller only ever saw the 401 — never the 2xx of the authenticated resend. Reset the latch when a fresh status line follows a delivered 401/407 (redirects never reach that path — their head is suppressed). - the prober's head callbacks return DataAction::abort to skip the body, which also aborts the transfer mid-handshake. Return `proceed` for a 401/407 when credentials were supplied, so curl's resend can run; the real status lands on the next header block. Also give ProbeRequest an `auth` field (default scheme == none) and pass it through base_request(), so a re-probe after a 401 can present the credentials the user just entered. No behaviour change when no auth is configured. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/include/vdm/net/probe.hpp | 1 + core/src/net/http_client.cpp | 7 +++++++ core/src/net/probe.cpp | 16 ++++++++++++++-- 3 files changed, 22 insertions(+), 2 deletions(-) diff --git a/core/include/vdm/net/probe.hpp b/core/include/vdm/net/probe.hpp index 2122190..5b1e425 100644 --- a/core/include/vdm/net/probe.hpp +++ b/core/include/vdm/net/probe.hpp @@ -35,6 +35,7 @@ struct ProbeRequest { std::string user_agent; std::string referrer; ProxyConfig proxy; + AuthConfig auth; // credentials for a re-probe after a 401 (leave scheme == none otherwise) long connect_timeout_ms = 15000; long overall_timeout_ms = 25000; // download.probe deadline is 30 s diff --git a/core/src/net/http_client.cpp b/core/src/net/http_client.cpp index 25f9575..a83b206 100644 --- a/core/src/net/http_client.cpp +++ b/core/src/net/http_client.cpp @@ -195,6 +195,13 @@ struct HttpClient::Impl { std::string_view line(buf, total); if (line.starts_with("HTTP/")) { + // A new status line after we already delivered a 401/407 means libcurl's + // CURLAUTH_ANY handshake just resent with credentials: let the head of this + // second response be delivered too, so callers see the real (2xx/4xx) status + // rather than the challenge. Redirects never reach here delivered — their head + // is suppressed below — so this only fires for the auth resend. + if (st->head_delivered && (st->line_status == 401 || st->line_status == 407)) + st->head_delivered = false; st->line_status = status_from_line(line); st->head.headers.clear(); // keep only the final response's headers return total; diff --git a/core/src/net/probe.cpp b/core/src/net/probe.cpp index 608e02e..0a2c2f2 100644 --- a/core/src/net/probe.cpp +++ b/core/src/net/probe.cpp @@ -134,6 +134,7 @@ struct Prober::Impl { r.user_agent = pr.user_agent; r.referrer = pr.referrer; r.proxy = pr.proxy; + r.auth = pr.auth; r.follow_redirects = true; r.accept_encoding = false; r.connect_timeout_ms = pr.connect_timeout_ms; @@ -155,7 +156,7 @@ struct Prober::Impl { TransferCallbacks cbs; cbs.on_head = [p](const ResponseHead &h) { absorb_head(p->result, h); - return DataAction::abort; + return head_action(p, h); }; cbs.on_data = [](ConstByteSpan) { return DataAction::abort; }; cbs.on_finished = [p](Result r) { on_head_done(p, std::move(r)); }; @@ -165,6 +166,17 @@ struct Prober::Impl { client_.start(std::move(req), std::move(cbs)); } + // We want no body from a probe, so the head callback normally aborts after headers. + // The exception: a 401/407 when we were handed credentials — libcurl's CURLAUTH_ANY + // has to see that response before it resends with Authorization, so let this one + // through (a HEAD has no body; the ranged GET's is a single byte). The final status + // then lands on the next header block. + static DataAction head_action(const std::shared_ptr

&p, const ResponseHead &h) { + if ((h.status == 401 || h.status == 407) && p->job.req.auth.scheme != AuthScheme::none) + return DataAction::proceed; + return DataAction::abort; + } + static void absorb_head(ProbeResult &res, const ResponseHead &h) { if (h.status) res.http_status = h.status; @@ -242,7 +254,7 @@ struct Prober::Impl { TransferCallbacks cbs; cbs.on_head = [p](const ResponseHead &h) { absorb_range_head(p->result, h); - return DataAction::abort; // we don't need the one body byte + return head_action(p, h); // abort after headers, except a 401/407 with creds }; cbs.on_data = [](ConstByteSpan) { return DataAction::abort; }; cbs.on_finished = [p](Result r) { on_range_done(p, std::move(r)); }; From efbf366c188368f4683e1d6b132a138a35d0ae70 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 20:19:11 +0400 Subject: [PATCH 3/4] core: add test-name filters to the vtest harness run_all() takes an optional substring list; vtest_main forwards argv and a comma-separated $VT_ONLY. No filter => run everything, as before. Makes iterating on one slow end-to-end case (the engine suite) practical without a framework swap. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/tests/support/vtest.hpp | 10 +++++++++- core/tests/support/vtest_main.cpp | 27 +++++++++++++++++++++++++-- 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/core/tests/support/vtest.hpp b/core/tests/support/vtest.hpp index fa48322..79b84a2 100644 --- a/core/tests/support/vtest.hpp +++ b/core/tests/support/vtest.hpp @@ -81,9 +81,17 @@ std::string show(const T &v) { } } -inline int run_all() { +inline int run_all(const std::vector &filters = {}) { int failed_cases = 0; for (const auto &c : registry()) { + if (!filters.empty()) { + bool match = false; + for (const auto &f : filters) + if (std::string_view(c.name).find(f) != std::string_view::npos) + match = true; + if (!match) + continue; + } int before = stats().failures; stats().current_fatal = false; std::fprintf(stderr, "[ RUN ] %s\n", c.name); diff --git a/core/tests/support/vtest_main.cpp b/core/tests/support/vtest_main.cpp index e80cce7..6c15fbf 100644 --- a/core/tests/support/vtest_main.cpp +++ b/core/tests/support/vtest_main.cpp @@ -1,6 +1,29 @@ // vtest_main.cpp — shared entry point for every CORE test binary. +#include +#include +#include + #include "vtest.hpp" -int main() { - return ::vt::run_all(); +// Optional filters: each argv argument (or a comma-separated entry in $VT_ONLY) is a +// substring; a test runs only if its name contains one of them. No filters => run all. +int main(int argc, char **argv) { + std::vector filters; + for (int i = 1; i < argc; ++i) + filters.emplace_back(argv[i]); + if (const char *env = std::getenv("VT_ONLY")) { + std::string cur; + for (const char *p = env;; ++p) { + if (*p == ',' || *p == '\0') { + if (!cur.empty()) + filters.push_back(cur); + cur.clear(); + if (*p == '\0') + break; + } else { + cur.push_back(*p); + } + } + } + return ::vt::run_all(filters); } From 91636f8a4daff3c5d8e92cce16aa89bac46c6756 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 20:19:31 +0400 Subject: [PATCH 4/4] =?UTF-8?q?core:=20add=20the=20download=20engine=20?= =?UTF-8?q?=E2=80=94=20task=20machine,=20DownloadHandle,=20Engine?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 8 of the CORE build order: the bodies behind the DownloadSpec / callback API reviewed in core/docs/engine-api-m1.md. Wires probe -> segment workers -> WriteBuffer -> SparseFile -> .veloxpart.meta -> retry/backoff -> SegmentBudget -> RateLimiter -> callbacks into one event-driven machine. - Engine (src/engine.cpp): owns HttpClient, Prober, SegmentBudget, RateLimiter and one timer jthread (min-heap of scheduled fns). start() builds a task and returns a DownloadHandle; ~Impl quiesces every task before joining the timer so no callback fires during teardown. - DownloadTaskState (src/task/download_task.cpp): one `mu` task lock; a shared_mutex over the worker map for the curl write path; callbacks collected under `mu` and fired after release via a separate deferred queue; weak_from_this() in every async hop. State machine over the CORE-owned EngineState subset, auto-pause on 401/407 and on a 200 where 206 was expected, validated resume via If-Range. - digest (src/task/digest.cpp): OpenSSL EVP hash_file() for the optional post-download checksum; links OpenSSL::Crypto PRIVATE. - Segmenter::release_segment(): hand a paused segment back to the pool unassigned so resume's assign_slot() picks it up instead of splitting a still-"assigned" range and orphaning its front half. - DownloadHandle now names the real control block (vdm::task:: DownloadTaskState, defined only in the engine TU) via a namespace-scope fwd decl and a public-but-effectively-engine-only ctor, replacing the nested State/friend pair. Every public signature is unchanged; DAEMON (vdm-79) confirmed sched/ names only the public API. Fixes found while building the end-to-end suite (tests/task/engine_test.cpp, 9 cases against tools/testserver, green under ASan/UBSan and TSan): - a dropped connection lost its unflushed WriteBuffer tail while advance() had already counted those bytes as done -> a retry resumed past an unwritten hole. Flush on the failure path. - when the byte counters hit total while other workers were still live, teardown dropped their buffered tails. Now: cancel them and let each worker's own seg_finished drain it (the `assembling` state), last one starts verification -- no cross-thread buffer access. - seg_head() let a 401 with credentials present abort before libcurl's resend; now it proceeds once and acts on the final status. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/CMakeLists.txt | 6 +- core/include/vdm/segment/segmenter.hpp | 4 + core/include/vdm/task/download.hpp | 13 +- core/src/engine.cpp | 169 ++++ core/src/segment/segmenter.cpp | 9 + core/src/task/digest.cpp | 89 ++ core/src/task/digest.hpp | 20 + core/src/task/download_task.cpp | 1178 ++++++++++++++++++++++++ core/src/task/download_task.hpp | 58 ++ core/tests/CMakeLists.txt | 7 + core/tests/task/engine_test.cpp | 328 +++++++ 11 files changed, 1876 insertions(+), 5 deletions(-) create mode 100644 core/src/engine.cpp create mode 100644 core/src/task/digest.cpp create mode 100644 core/src/task/digest.hpp create mode 100644 core/src/task/download_task.cpp create mode 100644 core/src/task/download_task.hpp create mode 100644 core/tests/task/engine_test.cpp diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt index fca1a84..8c365b8 100644 --- a/core/CMakeLists.txt +++ b/core/CMakeLists.txt @@ -6,6 +6,7 @@ find_package(Threads REQUIRED) find_package(CURL 8.0 REQUIRED) +find_package(OpenSSL REQUIRED) add_library(veloxcore STATIC src/util/error.cpp @@ -22,6 +23,9 @@ add_library(veloxcore STATIC src/meta/veloxpart.cpp src/segment/segmenter.cpp src/segment/budget.cpp + src/task/digest.cpp + src/task/download_task.cpp + src/engine.cpp ) add_library(velox::core ALIAS veloxcore) @@ -38,7 +42,7 @@ target_compile_options(veloxcore PRIVATE -Wall -Wextra -Wpedantic -Werror ) -target_link_libraries(veloxcore PUBLIC Threads::Threads CURL::libcurl) +target_link_libraries(veloxcore PUBLIC Threads::Threads CURL::libcurl PRIVATE OpenSSL::Crypto) # Later stages add: find_package(OpenSSL) for meta/ (streaming SHA-256 + resume CRC). diff --git a/core/include/vdm/segment/segmenter.hpp b/core/include/vdm/segment/segmenter.hpp index 1957ec1..3d322d5 100644 --- a/core/include/vdm/segment/segmenter.hpp +++ b/core/include/vdm/segment/segmenter.hpp @@ -131,6 +131,10 @@ class Segmenter { [[nodiscard]] SegState segment_state(std::uint32_t idx) const noexcept; void set_segment_state(std::uint32_t idx, SegState s) noexcept; + // Hand a segment back to the pool without touching its `completed`: it becomes an + // unassigned idle range that the next assign_slot() picks up (used on pause). + void release_segment(std::uint32_t idx) noexcept; + // Sum of bytes done across every segment (active + already complete). Locks. [[nodiscard]] std::uint64_t downloaded() const; [[nodiscard]] bool all_complete() const; diff --git a/core/include/vdm/task/download.hpp b/core/include/vdm/task/download.hpp index 4eb184b..4daa95c 100644 --- a/core/include/vdm/task/download.hpp +++ b/core/include/vdm/task/download.hpp @@ -31,6 +31,11 @@ class Engine; // owns and fills DownloadHandle (see vdm/engine.hpp) namespace vdm::task { +// The task control block. Opaque: defined only in the engine's translation unit. A handle +// holds a shared_ptr to one; the engine keeps its own copy so the task outlives a caller +// that drops its handle. +struct DownloadTaskState; + // --- input --------------------------------------------------------------------------- struct Checksum { @@ -177,6 +182,9 @@ struct DownloadCallbacks { class DownloadHandle { public: DownloadHandle() = default; + // The engine builds handles; `DownloadTaskState` is incomplete everywhere else, so + // this is effectively engine-only without a friend declaration. + explicit DownloadHandle(std::shared_ptr s) : state_(std::move(s)) {} [[nodiscard]] TaskId id() const noexcept; [[nodiscard]] bool valid() const noexcept { return static_cast(state_); } @@ -207,10 +215,7 @@ class DownloadHandle { [[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_; + std::shared_ptr state_; }; } // namespace vdm::task diff --git a/core/src/engine.cpp b/core/src/engine.cpp new file mode 100644 index 0000000..0ba4b68 --- /dev/null +++ b/core/src/engine.cpp @@ -0,0 +1,169 @@ +// vdm/engine.cpp + +#include "vdm/engine.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include "task/download_task.hpp" + +namespace vdm { + +struct Engine::Impl : task::TaskHost { + explicit Impl(Config c) + : cfg_(c), + http_(net::HttpClient::Options{.workers = c.http_workers}), + prober_(c.probe_pool_size ? c.probe_pool_size : 4), + budget_(segment::SegmentBudget::Options{.max_active_segments = c.max_active_segments}) { + timer_ = std::jthread([this](std::stop_token st) { timer_loop(st); }); + } + + ~Impl() override { + // Quiesce every task first so no callback fires during or after teardown: mark it + // retired and cancel its transfers. Then stop the timer thread (a fn already + // running holds a shared_ptr and finishes, but seg_finished early-returns on + // `retired`). Then drop the registry; http_/prober_/budget_ destruct after. + { + std::lock_guard lk(reg_mu_); + for (auto &[id, t] : tasks_) + task::quiesce_task(t); + } + timer_.request_stop(); + timer_cv_.notify_all(); + if (timer_.joinable()) + timer_.join(); + { + std::lock_guard lk(reg_mu_); + tasks_.clear(); + } + } + + // --- TaskHost ------------------------------------------------------------------- + net::HttpClient &http() override { return http_; } + segment::SegmentBudget &budget() override { return budget_; } + rate::RateLimiter &limiter() override { return limiter_; } + const Config &config() override { return cfg_; } + + task::TimerId schedule(SteadyTime at, std::function fn) override { + task::TimerId id; + { + std::lock_guard lk(timer_mu_); + id = ++next_timer_; + timers_.emplace(at, Entry{id, std::move(fn)}); + } + timer_cv_.notify_all(); + return id; + } + void cancel_timer(task::TimerId id) override { + std::lock_guard lk(timer_mu_); + for (auto it = timers_.begin(); it != timers_.end(); ++it) + if (it->second.id == id) { + timers_.erase(it); + return; + } + } + void probe(net::ProbeRequest req, std::function)> done) override { + prober_.probe(std::move(req), std::move(done)); + } + void task_retired(TaskId id) override { + std::lock_guard lk(reg_mu_); + tasks_.erase(id); + } + + // --- engine surface ---------------------------------------------------------------- + task::DownloadHandle start(task::DownloadSpec spec, task::DownloadCallbacks cbs) { + TaskId id{++next_id_}; + auto st = task::create_task(*this, id, std::move(spec), std::move(cbs)); + { + std::lock_guard lk(reg_mu_); + tasks_[id] = st; + } + return task::DownloadHandle(std::move(st)); + } + + Config cfg_; + net::HttpClient http_; + net::Prober prober_; + segment::SegmentBudget budget_; + rate::RateLimiter limiter_; + + std::atomic next_id_{0}; + + std::mutex reg_mu_; + std::unordered_map> tasks_; + + struct Entry { + task::TimerId id; + std::function fn; + }; + std::mutex timer_mu_; + std::condition_variable timer_cv_; + std::multimap timers_; + std::atomic next_timer_{0}; + std::jthread timer_; + + void timer_loop(std::stop_token st) { + std::unique_lock lk(timer_mu_); + while (!st.stop_requested()) { + if (timers_.empty()) { + timer_cv_.wait_for(lk, std::chrono::seconds(1)); + continue; + } + auto next_at = timers_.begin()->first; + if (timer_cv_.wait_until(lk, next_at, [&] { + return st.stop_requested() || + (!timers_.empty() && timers_.begin()->first < next_at); + })) { + continue; // stop, or an earlier timer landed — re-evaluate + } + // fire everything due + std::vector> due; + auto now = std::chrono::steady_clock::now(); + for (auto it = timers_.begin(); it != timers_.end() && it->first <= now;) + due.push_back(std::move(it->second.fn)), it = timers_.erase(it); + lk.unlock(); + for (auto &fn : due) + fn(); + lk.lock(); + } + } +}; + +// --- Engine --------------------------------------------------------------------------- + +Engine::Engine() : Engine(Config{}) {} +Engine::Engine(Config cfg) : impl_(std::make_unique(cfg)) {} +Engine::~Engine() = default; + +task::DownloadHandle Engine::start(task::DownloadSpec spec, task::DownloadCallbacks cbs) { + return impl_->start(std::move(spec), std::move(cbs)); +} + +segment::SegmentBudget &Engine::segment_budget() noexcept { + return impl_->budget_; +} +rate::RateLimiter &Engine::rate_limiter() noexcept { + return impl_->limiter_; +} + +void Engine::set_default_segments(std::uint32_t n) { + impl_->cfg_.default_segments = n ? n : 1; +} +void Engine::set_default_buffer_bytes(std::uint64_t b) { + impl_->cfg_.default_buffer_bytes = b; +} +void Engine::set_max_total_buffer_bytes(std::uint64_t b) { + impl_->cfg_.max_total_buffer_bytes = b; +} +void Engine::set_probe_pool_size(std::uint32_t) { /* prober pool is fixed at construction in M1 */ } + +void Engine::probe(net::ProbeRequest req, std::function)> done) { + impl_->prober_.probe(std::move(req), std::move(done)); +} + +} // namespace vdm diff --git a/core/src/segment/segmenter.cpp b/core/src/segment/segmenter.cpp index 2587024..15d98d7 100644 --- a/core/src/segment/segmenter.cpp +++ b/core/src/segment/segmenter.cpp @@ -263,6 +263,15 @@ void Segmenter::set_segment_state(std::uint32_t idx, SegState st) noexcept { segs_[idx].state.store(st); } +void Segmenter::release_segment(std::uint32_t idx) noexcept { + std::lock_guard lk(mu_); + if (idx >= segs_.size()) + return; + segs_[idx].assigned = false; + if (segs_[idx].state.load() != SegState::complete) + segs_[idx].state.store(SegState::idle); +} + // --- public: queries (take the lock) ---------------------------------------------- std::uint64_t Segmenter::downloaded() const { diff --git a/core/src/task/digest.cpp b/core/src/task/digest.cpp new file mode 100644 index 0000000..b9c6afe --- /dev/null +++ b/core/src/task/digest.cpp @@ -0,0 +1,89 @@ +// vdm/task/digest.cpp + +#include "task/digest.hpp" + +#include +#include + +#include +#include +#include + +#include + +namespace vdm::task { +namespace { + +const EVP_MD *md_for(Checksum::Algo a) { + switch (a) { + case Checksum::Algo::md5: + return EVP_md5(); + case Checksum::Algo::sha1: + return EVP_sha1(); + case Checksum::Algo::sha256: + return EVP_sha256(); + case Checksum::Algo::sha512: + return EVP_sha512(); + } + return EVP_sha256(); +} + +std::string to_hex(const unsigned char *p, unsigned n) { + static const char *h = "0123456789abcdef"; + std::string s; + s.reserve(n * 2); + for (unsigned i = 0; i < n; ++i) { + s.push_back(h[p[i] >> 4]); + s.push_back(h[p[i] & 0xF]); + } + return s; +} + +} // namespace + +Result hash_file(std::string_view path, Checksum::Algo algo) { + std::string p(path); + int fd = ::open(p.c_str(), O_RDONLY | O_CLOEXEC); + if (fd < 0) + return ErrorInfo(Error::path_rejected, + std::string("open ") + p + ": " + std::strerror(errno)); + + EVP_MD_CTX *ctx = EVP_MD_CTX_new(); + if (!ctx) { + ::close(fd); + return ErrorInfo(Error::internal, "EVP_MD_CTX_new"); + } + auto fail = [&](Error e, std::string msg) { + EVP_MD_CTX_free(ctx); + ::close(fd); + return Result(ErrorInfo(e, std::move(msg))); + }; + + if (EVP_DigestInit_ex(ctx, md_for(algo), nullptr) != 1) + return fail(Error::internal, "EVP_DigestInit_ex"); + + std::array buf{}; + for (;;) { + ssize_t n = ::read(fd, buf.data(), buf.size()); + if (n < 0) { + if (errno == EINTR) + continue; + return fail(Error::io_error, std::string("read: ") + std::strerror(errno)); + } + if (n == 0) + break; + if (EVP_DigestUpdate(ctx, buf.data(), static_cast(n)) != 1) + return fail(Error::internal, "EVP_DigestUpdate"); + } + + unsigned char out[EVP_MAX_MD_SIZE]; + unsigned out_len = 0; + if (EVP_DigestFinal_ex(ctx, out, &out_len) != 1) + return fail(Error::internal, "EVP_DigestFinal_ex"); + + EVP_MD_CTX_free(ctx); + ::close(fd); + return to_hex(out, out_len); +} + +} // namespace vdm::task diff --git a/core/src/task/digest.hpp b/core/src/task/digest.hpp new file mode 100644 index 0000000..16cd866 --- /dev/null +++ b/core/src/task/digest.hpp @@ -0,0 +1,20 @@ +// vdm/task/digest.hpp — internal: hash a finished file for checksum verification. + +#ifndef VDM_TASK_DIGEST_HPP +#define VDM_TASK_DIGEST_HPP + +#include +#include + +#include "vdm/task/download.hpp" +#include "vdm/util/result.hpp" + +namespace vdm::task { + +// Stream `path` through the digest and return it lower-case hex. io_error on a read +// failure, path_rejected if the file can't be opened. +[[nodiscard]] Result hash_file(std::string_view path, Checksum::Algo algo); + +} // namespace vdm::task + +#endif // VDM_TASK_DIGEST_HPP diff --git a/core/src/task/download_task.cpp b/core/src/task/download_task.cpp new file mode 100644 index 0000000..d67f3be --- /dev/null +++ b/core/src/task/download_task.cpp @@ -0,0 +1,1178 @@ +// vdm/task/download_task.cpp — the download state machine behind DownloadHandle. +// +// One std::mutex `mu` is the task lock: every state transition, worker start/stop, and +// terminal path runs under it. The curl write-path (seg_data / seg_head) touches only +// per-segment state, guarded by a shared_mutex over the worker map. Segment completions +// and timers are posted to the engine timer thread and run under `mu`. Callbacks to +// DAEMON are collected under `mu` (defer()) and fired only after it is released +// (flush_deferred(), which guards its own queue with a separate mutex). + +#include "task/download_task.hpp" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "task/digest.hpp" +#include "vdm/io/sparse_file.hpp" +#include "vdm/io/write_buffer.hpp" +#include "vdm/meta/veloxpart.hpp" +#include "vdm/net/url.hpp" +#include "vdm/segment/segmenter.hpp" + +namespace vdm::task { +namespace { + +constexpr std::uint64_t kBufFloor = 64u * 1024; +constexpr std::uint64_t kBufCeil = 16u * 1024 * 1024; +constexpr std::uint64_t kNoEnd = ~std::uint64_t{0}; + +bool is_connection_error(Error e) noexcept { + switch (e) { + case Error::connection_reset: + case Error::timeout: + case Error::connect_failed: + case Error::resolve_failed: + case Error::tls_failed: + return true; + default: + return false; + } +} + +std::chrono::milliseconds backoff_for(int attempt) { + // 1st retry ~1s, then 2, 4, 8, ... capped at 60s, +/-20% jitter (docs/04 §7). + int shift = std::clamp(attempt - 1, 0, 6); + std::uint64_t base = 1000ull << shift; + base = std::min(base, 60'000); + std::uint64_t jitter = base / 5; + std::uint64_t r = static_cast(std::rand()) % (2 * jitter + 1); + return std::chrono::milliseconds{base - jitter + r}; +} + +std::string lower(std::string s) { + std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); }); + return s; +} + +} // namespace + +struct SegWorker { + std::uint32_t seg_index = 0; + net::Transfer transfer; + std::unique_ptr buf; + std::uint64_t base_completed = 0; + std::uint64_t recv = 0; + + long http_status = 0; + bool needs_auth = false; + bool wrong_status = false; + bool range_bad = false; + bool auth_handshake = false; // saw a 401/407 and let libcurl resend with credentials + std::optional flush_error; + + int retries = 0; + + SteadyTime sample_at{}; + std::uint64_t sample_bytes = 0; + double speed_bps = 0; +}; + +struct DownloadTaskState : std::enable_shared_from_this { + TaskHost &host; + TaskId id; + DownloadSpec spec; + DownloadCallbacks cbs; + + std::mutex mu; + std::shared_mutex workers_mu; + std::mutex deferred_mu; + + EngineState state = EngineState::probing; + std::optional last_error; + std::atomic retired{false}; + + std::string part_path, meta_path; + + net::ProbeResult probe; + bool have_probe = false; + bool resumable = false; + bool probe_needed_auth = false; + std::optional total_size; + std::string origin_host; + + std::unique_ptr seg; + std::unique_ptr file; + std::unordered_map> workers; + std::unordered_map retry_counts; + std::uint32_t slot_target = 0; + std::uint32_t effective_buffer = 0; + std::uint32_t requested_segments = 8; + bool registered = false; + + bool pause_requested = false; + bool cancel_requested = false; + bool assembling = false; // every byte received; draining live workers' buffers to disk + bool discard_on_cancel = false; + bool awaiting_auth = false; + bool awaiting_decision = false; + int max_retries = 10; + + std::atomic last_progress_ns{0}; + SteadyTime started_at{}; + + std::vector> deferred; + + DownloadTaskState(TaskHost &h, TaskId i, DownloadSpec s, DownloadCallbacks c) + : host(h), id(i), spec(std::move(s)), cbs(std::move(c)) {} + + // --- deferred callbacks ------------------------------------------------------------- + void defer(std::function fn) { + std::lock_guard lk(deferred_mu); + deferred.push_back(std::move(fn)); + } + void flush_deferred() { // MUST be called with `mu` NOT held + std::vector> run; + { + std::lock_guard lk(deferred_mu); + run.swap(deferred); + } + for (auto &fn : run) + fn(); + } + + // caller holds mu + void transition(EngineState to, std::optional err) { + if (state == to && !err) + return; + EngineState from = state; + state = to; + last_error = err; + auto cb = cbs.on_state; + defer([cb, from, to, err] { + if (cb) + cb(from, to, err); + }); + } + [[nodiscard]] std::string current_url() const { + return have_probe && !probe.effective_url.empty() ? probe.effective_url : spec.url; + } + [[nodiscard]] bool has_mirror() const { return !spec.mirrors.empty(); } + [[nodiscard]] std::uint32_t want_slots() const { + if (!seg) + return requested_segments ? requested_segments : 1; + std::uint32_t target = seg->target_segment_count(); + std::uint32_t incomplete = 0; + bool any = false; + for (auto &v : seg->snapshot()) { + any = true; + if (v.state != segment::SegState::complete && v.state != segment::SegState::failed) + ++incomplete; + } + if (!any) + return target; // fresh: nothing created yet + if (incomplete == 0) + return target; // orphans pending re-split + return std::clamp(incomplete, 1, target); + } + + void begin(); + void on_probe_result(Result r); + void finish_probe_locked(); + void apply_slot_target(std::uint32_t n); + void start_worker_locked(std::uint32_t seg_idx); + void restart_probe(bool with_auth); + + net::DataAction seg_head(std::uint32_t seg_idx, const net::ResponseHead &h); + net::DataAction seg_data(std::uint32_t seg_idx, ConstByteSpan span); + void on_transfer_done(std::uint32_t seg_idx, Result r); + void seg_finished(std::uint32_t seg_idx, Result r); + void retry_worker(std::uint32_t seg_idx); + + void begin_verify_locked(); + void fail_locked(ErrorInfo e); + void auto_pause_locked(ErrorInfo e, bool auth, bool decision); + void cancel_all_transfers_locked(); + void start_assembly_locked(); + void finalize_cancel_locked(); + void write_sidecar_locked(); + void emit_progress_if_due(); + + void do_pause(); + void do_resume(); + void do_cancel(bool discard); + void do_provide_auth(std::string u, std::string p, bool remember); + void do_decide(Decision d); + void do_refresh_url(std::string url, std::vector headers); + EngineState snapshot_state(); + Progress snapshot_progress(); + void quiesce(); // engine shutdown: stop everything, fire nothing +}; + +// ================================================================================== + +void DownloadTaskState::begin() { + bool probe_hint; + { + std::unique_lock lk(mu); + part_path = spec.save_path + ".veloxpart"; + meta_path = part_path + ".meta"; + started_at = std::chrono::steady_clock::now(); + max_retries = + static_cast(spec.max_retries.value_or(host.config().default_max_retries)); + requested_segments = spec.segments.value_or(host.config().default_segments); + probe_hint = spec.probe_hint.has_value(); + if (probe_hint) { + probe = *spec.probe_hint; + have_probe = true; + finish_probe_locked(); + } else { + transition(EngineState::probing, std::nullopt); + } + } + flush_deferred(); + if (!probe_hint) + restart_probe(false); +} + +void DownloadTaskState::restart_probe(bool with_auth) { + net::ProbeRequest pr; + pr.url = spec.url; + pr.headers = spec.headers; + pr.cookies = spec.cookies; + pr.referrer = spec.referrer; + pr.user_agent = spec.user_agent; + pr.proxy = spec.proxy; + if (with_auth) + pr.auth = spec.auth; + auto wp = weak_from_this(); + host.probe(std::move(pr), [wp](Result r) { + if (auto s = wp.lock()) + s->on_probe_result(std::move(r)); + }); +} + +void DownloadTaskState::on_probe_result(Result r) { + { + std::unique_lock lk(mu); + if (retired.load() || is_terminal(state)) + return; + if (!r.has_value()) { + fail_locked(std::move(r).error()); + } else { + probe = std::move(r).value(); + have_probe = true; + if (probe.requires_auth) { + probe_needed_auth = true; + auto_pause_locked( + ErrorInfo(Error::auth_required, "probe 401/407", probe.http_status), true, + false); + } else { + finish_probe_locked(); + } + } + } + flush_deferred(); +} + +void DownloadTaskState::finish_probe_locked() { + resumable = probe.resumable; + total_size = probe.total_size; + origin_host = net::split_url(current_url()).host; + + const std::uint64_t min_seg = host.config().min_segment_bytes; + const std::uint64_t total = total_size.value_or(0); + + std::vector resumed; + bool do_resume = spec.allow_resume; + if (do_resume) { + auto m = meta::read_veloxpart_file(meta_path); + if (m.has_value() && total > 0 && m.value().total_size == total && + (m.value().etag.empty() || m.value().etag == probe.etag) && + (m.value().last_modified.empty() || m.value().last_modified == probe.last_modified)) { + for (auto &s : m.value().segments) + resumed.push_back({s.start, s.end, s.completed}); + } else { + do_resume = false; + } + } + + file = std::make_unique(); + io::SparseFile::OpenOptions oo; + oo.total_size = total; + oo.preallocate = total > 0; + oo.truncate_existing = !do_resume; + if (auto o = file->open(part_path, oo); !o.has_value()) { + fail_locked(std::move(o).error()); + return; + } + + if (do_resume && !resumed.empty()) + seg = std::make_unique(total, requested_segments, resumed, resumable, + min_seg); + else + seg = std::make_unique(total, requested_segments, resumable, min_seg); + + const std::uint32_t cap = resumable ? std::min(requested_segments, 32) : 1; + std::uint64_t want_buf = spec.buffer_bytes.value_or(host.config().default_buffer_bytes); + want_buf = std::clamp(want_buf, kBufFloor, kBufCeil); + std::uint64_t max_per = std::max( + kBufFloor, host.config().max_total_buffer_bytes / std::max(1u, cap)); + effective_buffer = static_cast(std::min(want_buf, max_per)); + + auto wp = weak_from_this(); + host.budget().register_task(id, {origin_host, cap, resumable}, [wp](std::uint32_t n) { + auto s = wp.lock(); + if (!s) + return; + s->host.schedule(std::chrono::steady_clock::now(), [wp, n] { + if (auto s2 = wp.lock()) + s2->apply_slot_target(n); + }); + }); + registered = true; + host.limiter().attach_task(id, std::nullopt); + + transition(EngineState::connecting, std::nullopt); + host.budget().set_want(id, want_slots()); +} + +void DownloadTaskState::apply_slot_target(std::uint32_t n) { + { + std::unique_lock lk(mu); + if (retired.load() || is_terminal(state) || pause_requested || cancel_requested || + awaiting_auth || awaiting_decision || assembling || !seg) + return; + slot_target = n; + while (workers.size() < slot_target) { + auto s = seg->assign_slot(); + if (!s) { + host.budget().set_want(id, static_cast(workers.size())); + break; + } + if (!host.budget().confirm_slot(id)) { + seg->set_segment_state(*s, segment::SegState::idle); + break; + } + start_worker_locked(*s); + } + if (state == EngineState::connecting && !workers.empty()) + transition(EngineState::downloading, std::nullopt); + } + flush_deferred(); +} + +void DownloadTaskState::start_worker_locked(std::uint32_t seg_idx) { + auto w = std::make_unique(); + w->seg_index = seg_idx; + const std::uint64_t sstart = seg->segment_start(seg_idx); + const std::uint64_t completed = seg->segment_completed(seg_idx); + const std::uint64_t send = seg->segment_end(seg_idx); + w->base_completed = completed; + w->sample_at = std::chrono::steady_clock::now(); + if (auto it = retry_counts.find(seg_idx); it != retry_counts.end()) + w->retries = it->second; + + auto wp = weak_from_this(); + w->buf = std::make_unique( + sstart + completed, effective_buffer, + [wp](std::uint64_t off, ConstByteSpan sp) -> Result { + if (auto s = wp.lock()) + return s->file->write_at(off, sp); + return ErrorInfo(Error::canceled, "task gone"); + }); + + net::Request req; + req.url = current_url(); + req.headers = spec.headers; + req.cookies = spec.cookies; + req.referrer = spec.referrer; + req.user_agent = spec.user_agent; + req.proxy = spec.proxy; + req.auth = spec.auth; + req.follow_redirects = true; + req.accept_encoding = false; + req.low_speed_bytes_per_sec = 1024; + req.low_speed_secs = 30; + // Only range-request a resumable source; a non-resumable single segment is a plain GET. + if (resumable && total_size && send != kNoEnd) + req.range = net::ByteRange{sstart + completed, send}; + if (completed > 0 && (!probe.etag.empty() || !probe.last_modified.empty())) + req.headers.push_back({"If-Range", !probe.etag.empty() ? probe.etag : probe.last_modified}); + + net::TransferCallbacks tc; + tc.on_head = [wp, seg_idx](const net::ResponseHead &h) { + if (auto s = wp.lock()) + return s->seg_head(seg_idx, h); + return net::DataAction::abort; + }; + tc.on_data = [wp, seg_idx](ConstByteSpan sp) { + if (auto s = wp.lock()) + return s->seg_data(seg_idx, sp); + return net::DataAction::abort; + }; + tc.on_finished = [wp, seg_idx](Result r) { + if (auto s = wp.lock()) + s->on_transfer_done(seg_idx, std::move(r)); + }; + + auto *raw = w.get(); + { + std::unique_lock wl(workers_mu); + workers[seg_idx] = std::move(w); + } + raw->transfer = host.http().start(std::move(req), std::move(tc)); + seg->set_segment_state(seg_idx, segment::SegState::connecting); +} + +// --- curl write path ------------------------------------------------------------------- + +net::DataAction DownloadTaskState::seg_head(std::uint32_t seg_idx, const net::ResponseHead &h) { + SegWorker *w = nullptr; + { + std::shared_lock lk(workers_mu); + auto it = workers.find(seg_idx); + if (it == workers.end()) + return net::DataAction::abort; + w = it->second.get(); + } + w->http_status = h.status; + if (h.status == 401 || h.status == 407) { + // libcurl with CURLAUTH_ANY does a challenge round-trip: the first response is a + // 401/407, then it resends with credentials. Don't kill that handshake — let it + // proceed once and wait for the real status on the next header block. Only when we + // have nothing to try, or we already tried, is this a genuine "auth required". + const bool have_creds = + spec.auth.scheme != net::AuthScheme::none && !spec.auth.username.empty(); + if (have_creds && !w->auth_handshake) { + w->auth_handshake = true; + return net::DataAction::proceed; + } + w->needs_auth = true; + return net::DataAction::abort; + } + // A 200 where we sent a Range is only "the file changed under us" when we are running + // a resumable, multi-segment transfer. A non-resumable single-segment GET legitimately + // gets a 200 (the source has no Range support). + if (resumable && total_size && *total_size > 0 && h.status == 200) { + w->wrong_status = true; + return net::DataAction::abort; + } + if (h.status == 416) { + w->range_bad = true; + return net::DataAction::abort; + } + if (h.status >= 400) + return net::DataAction::abort; + seg->set_segment_state(seg_idx, segment::SegState::downloading); + return net::DataAction::proceed; +} + +net::DataAction DownloadTaskState::seg_data(std::uint32_t seg_idx, ConstByteSpan span) { + SegWorker *w = nullptr; + { + std::shared_lock lk(workers_mu); + auto it = workers.find(seg_idx); + if (it == workers.end()) + return net::DataAction::abort; + w = it->second.get(); + } + if (span.empty()) + return net::DataAction::proceed; + // Body of libcurl's pre-auth 401/407 response — discard it; the real body follows the + // resend under a 2xx header. + if (w->http_status == 401 || w->http_status == 407) + return net::DataAction::proceed; + + auto wait = host.limiter().acquire(id, span.size()); + if (wait.count() > 0) { + net::Transfer t = w->transfer; + host.schedule(std::chrono::steady_clock::now() + wait, [t]() mutable { t.resume(); }); + return net::DataAction::pause; + } + + if (auto r = w->buf->append(span); !r.has_value()) { + w->flush_error = std::move(r).error(); + return net::DataAction::abort; + } + w->recv += span.size(); + seg->advance(seg_idx, w->base_completed + w->recv); + + auto now = std::chrono::steady_clock::now(); + auto dt = std::chrono::duration(now - w->sample_at).count(); + if (dt >= 0.5) { + double inst = static_cast(w->recv - w->sample_bytes) / dt; + w->speed_bps = w->speed_bps == 0 ? inst : 0.7 * w->speed_bps + 0.3 * inst; + w->sample_at = now; + w->sample_bytes = w->recv; + } + emit_progress_if_due(); + return net::DataAction::proceed; +} + +void DownloadTaskState::on_transfer_done(std::uint32_t seg_idx, Result r) { + auto wp = weak_from_this(); + host.schedule(std::chrono::steady_clock::now(), [wp, seg_idx, r = std::move(r)]() mutable { + if (auto s = wp.lock()) + s->seg_finished(seg_idx, std::move(r)); + }); +} + +void DownloadTaskState::seg_finished(std::uint32_t seg_idx, Result r) { + std::unique_lock lk(mu); + + std::unique_ptr w; + { + std::unique_lock wl(workers_mu); + auto it = workers.find(seg_idx); + if (it == workers.end()) { + lk.unlock(); + flush_deferred(); + return; + } + w = std::move(it->second); + workers.erase(it); + } + if (retired.load()) { // engine shutting down / already terminal — no more callbacks + if (w->buf) + (void)w->buf->flush(); + lk.unlock(); + return; + } + + // The bytes may all have arrived even though the connection closed dirty + // (content-length-mismatch's honest-length lie, flaky-reset's tail, a proxy RST after + // the last byte). If the segment is fully covered, that's a success. + if (seg && !cancel_requested && !pause_requested && !w->needs_auth && !w->wrong_status && + !w->flush_error && !w->range_bad) { + const std::uint64_t len = seg->segment_end(seg_idx) - seg->segment_start(seg_idx) + 1; + if (len != 0 && seg->segment_completed(seg_idx) >= len) { + r = Result(net::TransferStats{}); + } + } + auto release_slot = [&] { + if (registered) + host.budget().release_slot(id); + }; + auto done = [&] { + lk.unlock(); + flush_deferred(); + }; + + if (cancel_requested) { + if (w->buf) + (void)w->buf->flush(); + release_slot(); + if (workers.empty()) + finalize_cancel_locked(); + return done(); + } + if (pause_requested) { + if (w->buf) + (void)w->buf->flush(); + seg->advance(seg_idx, w->base_completed + w->recv); + // Hand the range back unassigned so resume's assign_slot() picks it up instead of + // treating it as still-held and splitting it (which would orphan its front half). + seg->release_segment(seg_idx); + release_slot(); + if (workers.empty()) { + (void)file->sync(); + write_sidecar_locked(); + transition(EngineState::paused, std::nullopt); + } + return done(); + } + if (assembling) { + // The file is fully received; this worker was cancelled so its buffered tail lands + // on disk. advance() has already counted these bytes; the flush makes them durable. + if (w->buf) { + if (auto f = w->buf->flush(); !f.has_value()) { + release_slot(); + fail_locked(std::move(f).error()); + return done(); + } + } + seg->advance(seg_idx, w->base_completed + w->recv); + release_slot(); + if (workers.empty()) + begin_verify_locked(); + return done(); + } + if (w->needs_auth) { + cancel_all_transfers_locked(); + release_slot(); + auto_pause_locked(ErrorInfo(Error::auth_required, "401/407", w->http_status), true, false); + return done(); + } + if (w->wrong_status) { + cancel_all_transfers_locked(); + release_slot(); + auto_pause_locked(ErrorInfo(Error::server_file_changed, "200 where 206 expected"), false, + true); + return done(); + } + if (w->flush_error) { + ErrorInfo e = *w->flush_error; + release_slot(); + if (e.code == Error::disk_full) { + cancel_all_transfers_locked(); + auto_pause_locked(e, false, false); + } else { + fail_locked(e); + } + return done(); + } + if (w->range_bad) + r = Result(ErrorInfo(Error::range_not_satisfiable, "416")); + + if (!r.has_value()) { + ErrorInfo e = std::move(r).error(); + // Persist what this attempt received before the connection dropped. seg_data() has + // already advanced `completed` past these bytes, but they are still only in the + // WriteBuffer (nothing forces a flush until the buffer fills or the segment ends). + // Without this, a retry resumes from `completed` and skips an unwritten hole — for + // a small file that never fills the buffer, every reset loses its whole payload. + if (w->buf) { + if (auto fl = w->buf->flush(); !fl.has_value()) { + release_slot(); + fail_locked(std::move(fl).error()); + return done(); + } + } + const bool conn = is_connection_error(e.code); + segment::FailAction fa = seg->on_failed(seg_idx, conn, has_mirror()); + if (w->recv > 0) + w->retries = 0; // progress this attempt: not a stuck segment + if (fa == segment::FailAction::requeue) { + retry_counts.erase(seg_idx); + release_slot(); + host.budget().set_want(id, want_slots()); + } else if (++w->retries > max_retries) { + ErrorInfo x(Error::max_retries_exhausted, e.to_string()); + x.cause = e.code; + release_slot(); + fail_locked(x); + } else { + retry_counts[seg_idx] = w->retries; + seg->set_segment_state(seg_idx, segment::SegState::stalled); + release_slot(); // give the slot back while we back off + int attempt = w->retries; + auto wp = weak_from_this(); + host.schedule(std::chrono::steady_clock::now() + backoff_for(attempt), [wp, seg_idx] { + if (auto s = wp.lock()) + s->retry_worker(seg_idx); + }); + if (workers.empty()) + transition(EngineState::retry_wait, std::nullopt); + } + return done(); + } + + // success + if (w->buf) { + if (auto f = w->buf->flush(); !f.has_value()) { + release_slot(); + fail_locked(std::move(f).error()); + return done(); + } + } + retry_counts.erase(seg_idx); + const bool may_steal = (workers.size() + 1) <= slot_target && !pause_requested; + auto cont = seg->on_complete(seg_idx, may_steal); + (void)file->sync(); + write_sidecar_locked(); + + if (cont) + start_worker_locked(*cont); // slot-neutral steal + else + release_slot(); + + if (seg->all_complete()) { + if (workers.empty()) { + begin_verify_locked(); + } else { + // Byte counters are satisfied, but other workers are still live and their + // tails may only be in their buffers. Cancel them; each one's seg_finished + // (this thread, once its curl worker has truly stopped) flushes via the + // `assembling` branch, and the last starts verification. + start_assembly_locked(); + } + } + return done(); +} + +void DownloadTaskState::retry_worker(std::uint32_t seg_idx) { + { + std::unique_lock lk(mu); + if (retired.load() || is_terminal(state) || pause_requested || cancel_requested || + assembling || !seg) + return; + if (workers.count(seg_idx)) + return; + if (!host.budget().confirm_slot(id)) + return; // budget is full; try again later + if (state == EngineState::retry_wait) + transition(EngineState::connecting, std::nullopt); + start_worker_locked(seg_idx); + if (!workers.empty() && state == EngineState::connecting) + transition(EngineState::downloading, std::nullopt); + } + flush_deferred(); +} + +void DownloadTaskState::begin_verify_locked() { + transition(EngineState::assembling, std::nullopt); + transition(EngineState::verifying, std::nullopt); + (void)file->sync(); + (void)file->close(); + + std::optional got; + if (spec.checksum) { + auto h = hash_file(part_path, spec.checksum->algo); + if (!h.has_value()) { + fail_locked(std::move(h).error()); + return; + } + got = h.value(); + if (*got != lower(spec.checksum->hex)) { + fail_locked(ErrorInfo(Error::checksum_mismatch, + "want " + lower(spec.checksum->hex) + " got " + *got)); + return; + } + } + if (::rename(part_path.c_str(), spec.save_path.c_str()) != 0) { + fail_locked(ErrorInfo(Error::io_error, std::string("rename: ") + std::strerror(errno))); + return; + } + ::unlink(meta_path.c_str()); + if (registered) + host.budget().deregister_task(id); + host.limiter().detach_task(id); + registered = false; + transition(EngineState::complete, std::nullopt); + + DownloadOutcome o; + o.final_path = spec.save_path; + o.bytes = seg->downloaded(); + o.sha256_hex = + (spec.checksum && spec.checksum->algo == Checksum::Algo::sha256) ? got : std::nullopt; + o.elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - started_at); + auto cb = cbs.on_finished; + defer([cb, o] { + if (cb) + cb(Result(o)); + }); + retired.store(true); + TaskHost *h = &host; + TaskId tid = id; + defer([h, tid] { h->task_retired(tid); }); +} + +void DownloadTaskState::fail_locked(ErrorInfo e) { + cancel_all_transfers_locked(); + if (file) + (void)file->close(); + if (seg) + write_sidecar_locked(); + if (registered) + host.budget().deregister_task(id); + host.limiter().detach_task(id); + registered = false; + transition(EngineState::failed, e); + auto cb = cbs.on_finished; + defer([cb, e] { + if (cb) + cb(Result(ErrorInfo(e))); + }); + retired.store(true); + TaskHost *h = &host; + TaskId tid = id; + defer([h, tid] { h->task_retired(tid); }); +} + +void DownloadTaskState::auto_pause_locked(ErrorInfo e, bool auth, bool decision) { + awaiting_auth = auth; + awaiting_decision = decision; + if (file) + (void)file->sync(); + if (seg) + write_sidecar_locked(); + if (registered) + host.budget().set_want(id, 0); + transition(EngineState::paused, e); + if (auth) { + AuthChallenge ac; + ac.host = origin_host; + auto cb = cbs.on_auth_required; + defer([cb, ac] { + if (cb) + cb(ac); + }); + } + if (decision) { + DecisionRequest dr; + dr.kind = e.code == Error::server_file_changed + ? DecisionRequest::Kind::server_file_changed + : DecisionRequest::Kind::range_metadata_stale; + dr.detail = e.context; + auto cb = cbs.on_decision_needed; + defer([cb, dr] { + if (cb) + cb(dr); + }); + } +} + +void DownloadTaskState::cancel_all_transfers_locked() { + std::unique_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); + workers.clear(); +} + +// Every byte is received but some workers are still live; their buffered tails would be +// lost if we dropped them here (seg_data() runs append() without workers_mu, so we cannot +// safely flush another segment's buffer from under it). Just cancel them and let each +// worker's own seg_finished drain it through the `assembling` branch once its curl worker +// has stopped. +void DownloadTaskState::start_assembly_locked() { + assembling = true; + transition(EngineState::assembling, std::nullopt); + std::shared_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); +} + +void DownloadTaskState::finalize_cancel_locked() { + if (file) + (void)file->close(); + if (registered) + host.budget().deregister_task(id); + host.limiter().detach_task(id); + registered = false; + if (discard_on_cancel) { + ::unlink(part_path.c_str()); + ::unlink(meta_path.c_str()); + } else if (seg) { + write_sidecar_locked(); + } + transition(EngineState::cancelled, std::nullopt); + auto cb = cbs.on_finished; + defer([cb] { + if (cb) + cb(Result(ErrorInfo(Error::canceled))); + }); + retired.store(true); + TaskHost *h = &host; + TaskId tid = id; + defer([h, tid] { h->task_retired(tid); }); +} + +void DownloadTaskState::write_sidecar_locked() { + if (!seg || !total_size) + return; + meta::VeloxPart vp; + vp.total_size = *total_size; + vp.urls = {spec.url}; + if (!current_url().empty() && current_url() != spec.url) + vp.urls.push_back(current_url()); + for (auto &m : spec.mirrors) + vp.urls.push_back(m); + vp.etag = probe.etag; + vp.last_modified = probe.last_modified; + vp.content_type = probe.mime; + std::uint64_t dl = 0; + for (auto &v : seg->snapshot()) { + if (v.state == segment::SegState::failed) + continue; + vp.segments.push_back({v.start, v.end, v.completed}); + dl += v.completed; + } + vp.downloaded = dl; + (void)meta::write_veloxpart_file(meta_path, vp, true); +} + +void DownloadTaskState::emit_progress_if_due() { + const auto now = std::chrono::steady_clock::now(); + const std::int64_t now_ns = now.time_since_epoch().count(); + std::int64_t prev = last_progress_ns.load(std::memory_order_relaxed); + if (now_ns - prev < 250'000'000) + return; + if (!last_progress_ns.compare_exchange_strong(prev, now_ns)) + return; + + Progress p; + { + std::shared_lock lk(workers_mu); + double agg = 0; + for (auto &[idx, w] : workers) { + agg += w->speed_bps; + SegmentProgress sp; + sp.index = idx; + sp.speed_bps = static_cast(w->speed_bps); + p.segments.push_back(sp); + } + p.speed_bps = static_cast(agg); + p.effective_segments = static_cast(workers.size()); + } + if (seg) { + p.downloaded = seg->downloaded(); + for (auto &v : seg->snapshot()) + for (auto &sp : p.segments) + if (sp.index == v.index) { + sp.start = v.start; + sp.end = v.end; + sp.completed = v.completed; + sp.state = v.state; + } + } + p.total = total_size; + p.effective_buffer_bytes = effective_buffer; + if (p.speed_bps > 0 && total_size && *total_size > p.downloaded) + p.eta_seconds = static_cast((*total_size - p.downloaded) / p.speed_bps); + auto cb = cbs.on_progress; + if (cb) + cb(p); +} + +// --- handle-facing -------------------------------------------------------------------- + +void DownloadTaskState::do_pause() { + { + std::unique_lock lk(mu); + if (state == EngineState::paused || is_terminal(state)) + return; + pause_requested = true; + if (registered) + host.budget().set_want(id, 0); + if (workers.empty()) { + if (file) + (void)file->sync(); + write_sidecar_locked(); + transition(EngineState::paused, std::nullopt); + } else { + std::shared_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); + } + } + flush_deferred(); +} + +void DownloadTaskState::do_resume() { + { + std::unique_lock lk(mu); + if (state != EngineState::paused || awaiting_auth || awaiting_decision) + return; + pause_requested = false; + transition(EngineState::connecting, std::nullopt); + if (registered) + host.budget().set_want(id, want_slots()); + } + flush_deferred(); +} + +void DownloadTaskState::do_cancel(bool discard) { + { + std::unique_lock lk(mu); + if (is_terminal(state)) + return; + cancel_requested = true; + discard_on_cancel = discard; + if (workers.empty()) { + finalize_cancel_locked(); + } else { + std::shared_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); + } + } + flush_deferred(); +} + +void DownloadTaskState::do_provide_auth(std::string u, std::string p, bool) { + bool reprobe = false; + { + std::unique_lock lk(mu); + if (!awaiting_auth) + return; + spec.auth = net::AuthConfig{net::AuthScheme::any, std::move(u), std::move(p)}; + awaiting_auth = false; + reprobe = probe_needed_auth; + probe_needed_auth = false; + transition(EngineState::connecting, std::nullopt); + if (!reprobe && registered) + host.budget().set_want(id, want_slots()); + } + flush_deferred(); + if (reprobe) + restart_probe(true); +} + +void DownloadTaskState::do_decide(Decision d) { + { + std::unique_lock lk(mu); + if (!awaiting_decision) + return; + awaiting_decision = false; + if (d == Decision::abort) { + fail_locked(ErrorInfo(Error::server_file_changed, "user aborted")); + } else { + if (d == Decision::restart) { + ::unlink(part_path.c_str()); + ::unlink(meta_path.c_str()); + spec.allow_resume = false; + io::SparseFile::OpenOptions oo; + oo.total_size = total_size.value_or(0); + oo.preallocate = oo.total_size > 0; + oo.truncate_existing = true; + file = std::make_unique(); + (void)file->open(part_path, oo); + seg = std::make_unique(total_size.value_or(0), + requested_segments, resumable, + host.config().min_segment_bytes); + } + transition(EngineState::connecting, std::nullopt); + if (registered) + host.budget().set_want(id, want_slots()); + } + } + flush_deferred(); +} + +void DownloadTaskState::do_refresh_url(std::string url, std::vector headers) { + { + std::unique_lock lk(mu); + if (is_terminal(state)) + return; + spec.url = std::move(url); + if (!headers.empty()) + spec.headers = std::move(headers); + std::shared_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); + } + flush_deferred(); + auto wp = weak_from_this(); + net::ProbeRequest pr; + pr.url = spec.url; + pr.headers = spec.headers; + pr.auth = spec.auth; + pr.proxy = spec.proxy; + host.probe(std::move(pr), [wp](Result r) { + auto s = wp.lock(); + if (!s) + return; + std::unique_lock lk(s->mu); + if (s->retired.load() || is_terminal(s->state)) + return; + if (r.has_value()) { + s->probe.effective_url = r.value().effective_url; + s->probe.etag = r.value().etag; + s->probe.last_modified = r.value().last_modified; + } + if (s->registered) + s->host.budget().set_want(s->id, s->want_slots()); + lk.unlock(); + s->flush_deferred(); + }); +} + +void DownloadTaskState::quiesce() { + std::lock_guard lk(mu); + retired.store(true); + std::unique_lock wl(workers_mu); + for (auto &[idx, w] : workers) + w->transfer.cancel(); + workers.clear(); +} + +EngineState DownloadTaskState::snapshot_state() { + std::lock_guard lk(mu); + return state; +} + +Progress DownloadTaskState::snapshot_progress() { + Progress p; + std::lock_guard lk(mu); + if (seg) { + p.downloaded = seg->downloaded(); + for (auto &v : seg->snapshot()) { + SegmentProgress sp; + sp.index = v.index; + sp.start = v.start; + sp.end = v.end; + sp.completed = v.completed; + sp.state = v.state; + p.segments.push_back(sp); + } + } + p.total = total_size; + { + std::shared_lock wl(workers_mu); + p.effective_segments = static_cast(workers.size()); + } + p.effective_buffer_bytes = effective_buffer; + return p; +} + +// ================================================================================== + +std::shared_ptr create_task(TaskHost &host, TaskId id, DownloadSpec spec, + DownloadCallbacks cbs) { + auto s = std::make_shared(host, id, std::move(spec), std::move(cbs)); + s->begin(); + return s; +} + +void quiesce_task(const std::shared_ptr &s) { + if (s) + s->quiesce(); +} + +// --- DownloadHandle bodies ---------------------------------------------------------- + +TaskId DownloadHandle::id() const noexcept { + return state_ ? state_->id : TaskId{}; +} +void DownloadHandle::pause() { + if (state_) + state_->do_pause(); +} +void DownloadHandle::resume() { + if (state_) + state_->do_resume(); +} +void DownloadHandle::cancel(bool discard_partial) { + if (state_) + state_->do_cancel(discard_partial); +} +void DownloadHandle::provide_auth(std::string u, std::string p, bool remember) { + if (state_) + state_->do_provide_auth(std::move(u), std::move(p), remember); +} +void DownloadHandle::decide(Decision d) { + if (state_) + state_->do_decide(d); +} +void DownloadHandle::refresh_url(std::string url, std::vector headers) { + if (state_) + state_->do_refresh_url(std::move(url), std::move(headers)); +} +EngineState DownloadHandle::state() const { + return state_ ? state_->snapshot_state() : EngineState::failed; +} +Progress DownloadHandle::progress() const { + return state_ ? state_->snapshot_progress() : Progress{}; +} + +} // namespace vdm::task diff --git a/core/src/task/download_task.hpp b/core/src/task/download_task.hpp new file mode 100644 index 0000000..07d5fe9 --- /dev/null +++ b/core/src/task/download_task.hpp @@ -0,0 +1,58 @@ +// vdm/task/download_task.hpp — internal: the task machine behind DownloadHandle, and the +// narrow interface it uses to reach engine-owned resources (so this TU doesn't depend on +// Engine::Impl). + +#ifndef VDM_TASK_DOWNLOAD_TASK_HPP +#define VDM_TASK_DOWNLOAD_TASK_HPP + +#include +#include +#include + +#include "vdm/engine.hpp" +#include "vdm/ids.hpp" +#include "vdm/net/http_client.hpp" +#include "vdm/net/probe.hpp" +#include "vdm/rate/token_bucket.hpp" +#include "vdm/segment/budget.hpp" +#include "vdm/task/download.hpp" + +namespace vdm::task { + +using TimerId = std::uint64_t; + +// Implemented by Engine::Impl. Every method is safe to call from any thread. +struct TaskHost { + virtual ~TaskHost() = default; + + virtual net::HttpClient &http() = 0; + virtual segment::SegmentBudget &budget() = 0; + virtual rate::RateLimiter &limiter() = 0; + virtual const Engine::Config &config() = 0; + + // One-shot timer. `fn` runs on the engine's timer thread. cancel_timer is a no-op if + // it already fired or never existed. + virtual TimerId schedule(SteadyTime at, std::function fn) = 0; + virtual void cancel_timer(TimerId id) = 0; + + // Probe pool, outside the segment budget (ADR 0011 §5). + virtual void probe(net::ProbeRequest req, + std::function)> done) = 0; + + // The task reached a terminal state — drop it from the engine's registry. + virtual void task_retired(TaskId id) = 0; +}; + +// Create a task and begin it (probe or connect). The returned control block is what +// DownloadHandle wraps (DownloadHandle{state}); the engine keeps its own copy so the task +// outlives a caller that drops its handle. +[[nodiscard]] std::shared_ptr create_task(TaskHost &host, TaskId id, + DownloadSpec spec, + DownloadCallbacks cbs); + +// Engine shutdown: stop every transfer and fire no further callbacks. Safe on nullptr. +void quiesce_task(const std::shared_ptr &s); + +} // namespace vdm::task + +#endif // VDM_TASK_DOWNLOAD_TASK_HPP diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index ca11711..d572ad8 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -34,6 +34,13 @@ 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) +vdm_add_test(veloxcore_engine_test task/engine_test.cpp) +target_include_directories(veloxcore_engine_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/net ${CMAKE_SOURCE_DIR}/core/src) +if(EXISTS ${_testserver}) + target_compile_definitions(veloxcore_engine_test PRIVATE VDM_TESTSERVER_PY="${_testserver}") + set_tests_properties(veloxcore_engine_test PROPERTIES TIMEOUT 300) +endif() + foreach(net_it http_client probe) vdm_add_test(veloxcore_${net_it}_test net/${net_it}_test.cpp) target_include_directories(veloxcore_${net_it}_test diff --git a/core/tests/task/engine_test.cpp b/core/tests/task/engine_test.cpp new file mode 100644 index 0000000..8e6127a --- /dev/null +++ b/core/tests/task/engine_test.cpp @@ -0,0 +1,328 @@ +// End-to-end: a real Engine against tools/testserver, covering the CORE M1 DoD paths. + +#include "vdm/engine.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "task/digest.hpp" +#include "testserver_fixture.hpp" +#include "vtest.hpp" + +using namespace vdm; +using namespace vdm::task; +using vdm::testing::TestServer; +using namespace std::chrono_literals; + +namespace { + +struct TmpDir { + std::string path; + TmpDir() { + const char *d = std::getenv("TMPDIR"); + path = (d ? d : "/tmp"); + path += "/vdm_engine_XXXXXX"; + path = ::mkdtemp(path.data()) ? path : ""; + } + ~TmpDir() { + // best-effort recursive cleanup of our flat dir + if (path.empty()) + return; + std::string cmd = "rm -rf '" + path + "'"; + (void)std::system(cmd.c_str()); + } + std::string file(const std::string &name) const { return path + "/" + name; } +}; + +struct Recorder { + std::promise> done; + std::future> fut = done.get_future(); + std::atomic fired{false}; + std::vector states; + std::mutex mu; + std::atomic auth_calls{0}; + std::atomic decision_calls{0}; + // A probe callback can fire before the caller has stored the handle returned by + // eng.start(). Callbacks that reach back into the handle wait on this. + std::atomic handle_ready{false}; + + void arm(DownloadHandle &) { handle_ready.store(true, std::memory_order_release); } + + DownloadCallbacks cbs(DownloadHandle *h = nullptr, std::string user = "", + std::string pass = "") { + DownloadCallbacks c; + c.on_state = [this](EngineState, EngineState to, const std::optional &) { + std::lock_guard lk(mu); + states.push_back(to); + }; + c.on_finished = [this](Result r) { + if (!fired.exchange(true)) + done.set_value(std::move(r)); + }; + if (h) { + c.on_auth_required = [this, h, user, pass](const AuthChallenge &) { + auth_calls.fetch_add(1); + while (!handle_ready.load(std::memory_order_acquire)) + std::this_thread::sleep_for(1ms); + h->provide_auth(user, pass, false); + }; + } + return c; + } + Result wait(std::chrono::seconds to = 40s) { + if (fut.wait_for(to) != std::future_status::ready) + return Err{Error::timeout, "engine test wait"}; + return fut.get(); + } + bool saw(EngineState s) { + std::lock_guard lk(mu); + for (auto x : states) + if (x == s) + return true; + return false; + } +}; + +std::uint64_t file_size(const std::string &p) { + int fd = ::open(p.c_str(), O_RDONLY); + if (fd < 0) + return ~0ull; + off_t e = ::lseek(fd, 0, SEEK_END); + ::close(fd); + return e < 0 ? ~0ull : static_cast(e); +} + +// The reference SHA-256 the testserver will report for a given path. +std::string server_sha(TestServer &srv, const std::string &mode, const std::string &size) { + // one-shot GET //sha256/ via a throwaway Engine::probe? no — use raw curl + // through a small helper. Simplest: shell out. + std::string url = srv.url("/" + mode + "/sha256/" + size); + std::string cmd = "curl -s '" + url + "'"; + std::string out; + if (FILE *f = ::popen(cmd.c_str(), "r")) { + char buf[512]; + while (std::fgets(buf, sizeof buf, f)) + out += buf; + ::pclose(f); + } + auto q = out.find("\"sha256\""); + if (q == std::string::npos) + return {}; + auto colon = out.find(':', q); + auto open = out.find('"', colon); + auto close = out.find('"', open + 1); + if (open == std::string::npos || close == std::string::npos) + return {}; + return out.substr(open + 1, close - open - 1); +} + +DownloadSpec spec_for(TestServer &srv, const std::string &urlpath, const std::string &save) { + DownloadSpec s; + s.url = srv.url(urlpath); + s.save_path = save; + return s; +} + +} // namespace + +VT_TEST(engine_plain_multisegment_download) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + VT_REQUIRE(!td.path.empty()); + + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/plain/file/4M", td.file("a.bin")), rec.cbs()); + + auto r = rec.wait(); + VT_REQUIRE(r.has_value()); + VT_CHECK_EQ(r.value().final_path, td.file("a.bin")); + VT_CHECK_EQ(r.value().bytes, 4u * 1024 * 1024); + VT_CHECK_EQ(file_size(td.file("a.bin")), 4u * 1024 * 1024); + VT_CHECK(rec.saw(EngineState::downloading)); + VT_CHECK(rec.saw(EngineState::complete)); + + auto got = hash_file(td.file("a.bin"), Checksum::Algo::sha256); + VT_REQUIRE(got.has_value()); + VT_CHECK_EQ(got.value(), server_sha(srv, "plain", "4M")); + // the sidecar is gone on success + VT_CHECK_EQ(::access((td.file("a.bin") + ".veloxpart.meta").c_str(), F_OK), -1); +} + +VT_TEST(engine_checksum_pass_and_mismatch) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Engine eng; + + std::string want = server_sha(srv, "plain", "1M"); + VT_REQUIRE(!want.empty()); + + { + Recorder rec; + auto s = spec_for(srv, "/plain/file/1M", td.file("ok.bin")); + s.checksum = Checksum{Checksum::Algo::sha256, want}; + auto h = eng.start(std::move(s), rec.cbs()); + auto r = rec.wait(); + VT_REQUIRE(r.has_value()); + VT_CHECK(rec.saw(EngineState::verifying)); + } + { + Recorder rec; + auto s = spec_for(srv, "/plain/file/1M", td.file("bad.bin")); + s.checksum = Checksum{Checksum::Algo::sha256, std::string(64, 'a')}; + auto h = eng.start(std::move(s), rec.cbs()); + auto r = rec.wait(); + VT_REQUIRE(!r.has_value()); + VT_CHECK_EQ(r.error().code, Error::checksum_mismatch); + } +} + +VT_TEST(engine_non_resumable_single_segment) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/no-range/file/2M", td.file("nr.bin")), rec.cbs()); + auto r = rec.wait(); + VT_REQUIRE(r.has_value()); + VT_CHECK_EQ(file_size(td.file("nr.bin")), 2u * 1024 * 1024); + auto got = hash_file(td.file("nr.bin"), Checksum::Algo::sha256); + VT_CHECK_EQ(got.value(), server_sha(srv, "no-range", "2M")); +} + +VT_TEST(engine_404_is_an_error) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/plain/nope", td.file("x.bin")), rec.cbs()); + auto r = rec.wait(); + VT_REQUIRE(!r.has_value()); + VT_CHECK_EQ(r.error().code, Error::not_found); + VT_CHECK(rec.saw(EngineState::failed)); +} + +VT_TEST(engine_cancel_mid_download) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/throttled/file/8M", td.file("c.bin")), rec.cbs()); + + for (int i = 0; i < 200 && !rec.saw(EngineState::downloading); ++i) + std::this_thread::sleep_for(10ms); + VT_REQUIRE(rec.saw(EngineState::downloading)); + h.cancel(/*discard_partial=*/true); + + auto r = rec.wait(); + VT_REQUIRE(!r.has_value()); + VT_CHECK_EQ(r.error().code, Error::canceled); + VT_CHECK(rec.saw(EngineState::cancelled)); + VT_CHECK_EQ(::access((td.file("c.bin") + ".veloxpart").c_str(), F_OK), -1); // discarded +} + +VT_TEST(engine_pause_resume_completes) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/throttled/file/2M", td.file("pr.bin")), rec.cbs()); + + for (int i = 0; i < 200 && !rec.saw(EngineState::downloading); ++i) + std::this_thread::sleep_for(10ms); + h.pause(); + for (int i = 0; i < 100 && h.state() != EngineState::paused; ++i) + std::this_thread::sleep_for(20ms); + VT_CHECK_EQ(h.state(), EngineState::paused); + h.resume(); + + auto r = rec.wait(90s); + VT_REQUIRE(r.has_value()); + VT_CHECK_EQ(file_size(td.file("pr.bin")), 2u * 1024 * 1024); + auto got = hash_file(td.file("pr.bin"), Checksum::Algo::sha256); + VT_CHECK_EQ(got.value(), server_sha(srv, "throttled", "2M")); +} + +VT_TEST(engine_resume_after_a_fresh_task) { + // Simulates kill -9: cancel WITHOUT discard, then a new task with allow_resume picks + // up the .veloxpart[.meta] and finishes with a byte-identical file. + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + std::string save = td.file("resume.bin"); + std::string want = server_sha(srv, "throttled", "3M"); + + { + Recorder rec; + Engine eng; + auto h = eng.start(spec_for(srv, "/throttled/file/3M", save), rec.cbs()); + for (int i = 0; i < 800 && (h.progress().downloaded < 512u * 1024); ++i) + std::this_thread::sleep_for(10ms); + VT_REQUIRE(h.progress().downloaded >= 512u * 1024); + h.cancel(/*discard_partial=*/false); + (void)rec.wait(); + VT_CHECK_EQ(::access((save + ".veloxpart.meta").c_str(), F_OK), 0); // sidecar kept + } + { + Recorder rec; + Engine eng; + auto s = spec_for(srv, "/throttled/file/3M", save); // same source -> sidecar validates + s.allow_resume = true; + auto h = eng.start(std::move(s), rec.cbs()); + auto r = rec.wait(120s); + VT_REQUIRE(r.has_value()); + VT_CHECK_EQ(file_size(save), 3u * 1024 * 1024); + auto got = hash_file(save, Checksum::Algo::sha256); + VT_CHECK_EQ(got.value(), want); // byte-identical after resume + } +} + +VT_TEST(engine_flaky_reset_retries_to_completion) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + // flaky-reset RSTs the first two attempts per (path,range); a single-segment request + // therefore needs the retry loop. + auto s = spec_for(srv, "/flaky-reset/file/256K", td.file("fl.bin")); + s.segments = 1; + s.max_retries = 40; // the server RSTs at the halfway point every attempt -> ~18 halvings + auto h = eng.start(std::move(s), rec.cbs()); + auto r = rec.wait(120s); + VT_REQUIRE(r.has_value()); + VT_CHECK_EQ(file_size(td.file("fl.bin")), 256u * 1024); + auto got = hash_file(td.file("fl.bin"), Checksum::Algo::sha256); + VT_CHECK_EQ(got.value(), server_sha(srv, "flaky-reset", "256K")); + VT_CHECK(rec.saw(EngineState::retry_wait) || rec.saw(EngineState::connecting)); +} + +VT_TEST(engine_401_then_provide_auth_completes) { + TestServer srv; + VT_REQUIRE(srv.available()); + TmpDir td; + Recorder rec; + Engine eng; + DownloadHandle h; + auto cbs = rec.cbs(&h, "test", "test"); + h = eng.start(spec_for(srv, "/401-basic/file/1M", td.file("au.bin")), std::move(cbs)); + rec.arm(h); // publish h to the auth callback (which may already be waiting) + + auto r = rec.wait(); + VT_REQUIRE(r.has_value()); + VT_CHECK(rec.auth_calls.load() >= 1); + VT_CHECK_EQ(file_size(td.file("au.bin")), 1u * 1024 * 1024); +}