core: add the download engine — task machine, DownloadHandle, Engine

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 <[email protected]>
Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS

core/src/engine.cpp

@@ -0,0 +1,169 @@
+// vdm/engine.cpp
+
+#include "vdm/engine.hpp"
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <map>
+#include <mutex>
+#include <thread>
+#include <unordered_map>
+
+#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<void()> 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<void(Result<net::ProbeResult>)> 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<std::uint64_t> next_id_{0};
+
+    std::mutex reg_mu_;
+    std::unordered_map<TaskId, std::shared_ptr<task::DownloadTaskState>> tasks_;
+
+    struct Entry {
+        task::TimerId id;
+        std::function<void()> fn;
+    };
+    std::mutex timer_mu_;
+    std::condition_variable timer_cv_;
+    std::multimap<SteadyTime, Entry> timers_;
+    std::atomic<std::uint64_t> 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<std::function<void()>> 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<Impl>(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<void(Result<net::ProbeResult>)> done) {
+    impl_->prober_.probe(std::move(req), std::move(done));
+}
+
+}  // namespace vdm

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 {

core/src/task/digest.cpp

@@ -0,0 +1,89 @@
+// vdm/task/digest.cpp
+
+#include "task/digest.hpp"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <array>
+#include <cerrno>
+#include <cstring>
+
+#include <openssl/evp.h>
+
+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<std::string> 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<std::string>(ErrorInfo(e, std::move(msg)));
+    };
+
+    if (EVP_DigestInit_ex(ctx, md_for(algo), nullptr) != 1)
+        return fail(Error::internal, "EVP_DigestInit_ex");
+
+    std::array<unsigned char, 256 * 1024> 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<std::size_t>(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

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 <string>
+#include <string_view>
+
+#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<std::string> hash_file(std::string_view path, Checksum::Algo algo);
+
+}  // namespace vdm::task
+
+#endif  // VDM_TASK_DIGEST_HPP

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 <cstdint>
+#include <functional>
+#include <memory>
+
+#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<void()> 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<void(Result<net::ProbeResult>)> 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<DownloadTaskState> 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<DownloadTaskState> &s);
+
+}  // namespace vdm::task
+
+#endif  // VDM_TASK_DOWNLOAD_TASK_HPP