core: add util layer — Result, Error taxonomy, bytes, event bus, pool, log

util/ carries no wire surface, so it lands before the contract freeze.

- error: enum class Error, the engine-wide failure taxonomy; is_retryable
  enumerates every value (no default:) so -Wswitch forces the retry
  decision on each future addition. ErrorInfo carries context/http_status.
- result: Result<T> over std::expected<T, ErrorInfo>, Result<void>,
  VDM_TRY / VDM_TRY_ASSIGN. Errors returned, never thrown, on the
  transfer path.
- bytes: span aliases, LE load_le/store_le (debug-asserted precondition,
  not input validation), and a bounds-checked latching ByteReader for the
  .veloxpart.meta reader.
- event_bus: typed thread-safe pub/sub; header states plainly that
  unsubscribe is not a quiesce point and download_task will need its own
  drain.
- thread_pool: std::jthread pool; dtor joins in the body before members
  die (fixed a use-after-destruction on cv_/mu_). Header notes shutdown is
  drain-only and DAEMON will need a cancel mode.
- log: sink interface (core does no I/O); DAEMON installs one.

Tested: -Werror clean, 6 binaries green under plain / ASan+UBSan / TSan.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS

core/tests/util/bytes_test.cpp

@@ -0,0 +1,94 @@
+#include "vdm/util/bytes.hpp"
+
+#include <array>
+#include <cstdint>
+#include <cstring>
+#include <string>
+#include <string_view>
+
+#include "vtest.hpp"
+
+using vdm::as_bytes;
+using vdm::as_chars;
+using vdm::ByteReader;
+using vdm::ByteSpan;
+using vdm::ConstByteSpan;
+using vdm::load_le;
+using vdm::store_le;
+
+VT_TEST(le_roundtrip_u32) {
+    std::array<std::byte, 4> buf{};
+    store_le<std::uint32_t>(buf, 0x01020304u);
+    // little-endian: least significant byte first
+    VT_CHECK_EQ(std::to_integer<int>(buf[0]), 0x04);
+    VT_CHECK_EQ(std::to_integer<int>(buf[1]), 0x03);
+    VT_CHECK_EQ(std::to_integer<int>(buf[2]), 0x02);
+    VT_CHECK_EQ(std::to_integer<int>(buf[3]), 0x01);
+    VT_CHECK_EQ(load_le<std::uint32_t>(buf), 0x01020304u);
+}
+
+VT_TEST(le_roundtrip_u64) {
+    std::array<std::byte, 8> buf{};
+    const std::uint64_t v = 0xDEADBEEF0BADF00Dull;
+    store_le<std::uint64_t>(buf, v);
+    VT_CHECK_EQ(load_le<std::uint64_t>(buf), v);
+}
+
+VT_TEST(as_bytes_as_chars_roundtrip) {
+    std::string_view s = "veloxpart";
+    ConstByteSpan b = as_bytes(s);
+    VT_CHECK_EQ(b.size(), s.size());
+    VT_CHECK_EQ(std::string(as_chars(b)), std::string(s));
+}
+
+VT_TEST(byte_reader_sequential_reads) {
+    // magic "VDMP", u16 version=2, u16 flags=0, u64 total=5
+    std::array<std::byte, 4 + 2 + 2 + 8> buf{};
+    ByteSpan w(buf);
+    std::memcpy(buf.data(), "VDMP", 4);
+    store_le<std::uint16_t>(w.subspan(4), 2);
+    store_le<std::uint16_t>(w.subspan(6), 0);
+    store_le<std::uint64_t>(w.subspan(8), 5);
+
+    ByteReader r{ConstByteSpan(buf)};
+    VT_CHECK_EQ(std::string(as_chars(r.raw(4))), std::string("VDMP"));
+    VT_CHECK_EQ(r.u16(), 2);
+    VT_CHECK_EQ(r.u16(), 0);
+    VT_CHECK_EQ(r.u64(), 5u);
+    VT_CHECK(!r.overran());
+    VT_CHECK_EQ(r.remaining(), 0u);
+}
+
+VT_TEST(byte_reader_latches_on_overrun) {
+    std::array<std::byte, 3> buf{};
+    ByteReader r{ConstByteSpan(buf)};
+    VT_CHECK_EQ(r.u16(), 0);  // ok, 2 of 3 consumed
+    VT_CHECK(!r.overran());
+    VT_CHECK_EQ(r.u32(), 0u);  // wants 4, only 1 left -> latch
+    VT_CHECK(r.overran());
+    VT_CHECK_EQ(r.remaining(), 0u);
+    // further reads stay zero and latched
+    VT_CHECK_EQ(r.u8(), 0);
+    VT_CHECK(r.overran());
+}
+
+VT_TEST(byte_reader_length_prefixed_string) {
+    std::array<std::byte, 4 + 5> buf{};
+    ByteSpan w(buf);
+    store_le<std::uint32_t>(w, 5);
+    std::memcpy(buf.data() + 4, "hello", 5);
+
+    ByteReader r{ConstByteSpan(buf)};
+    VT_CHECK_EQ(std::string(r.lp_string()), std::string("hello"));
+    VT_CHECK(!r.overran());
+}
+
+VT_TEST(byte_reader_length_prefixed_string_rejects_bogus_length) {
+    std::array<std::byte, 4 + 2> buf{};
+    ByteSpan w(buf);
+    store_le<std::uint32_t>(w, 0xFFFFFFFFu);  // claims 4 GiB, only 2 bytes follow
+    ByteReader r{ConstByteSpan(buf)};
+    std::string_view s = r.lp_string();
+    VT_CHECK(s.empty());
+    VT_CHECK(r.overran());
+}

core/tests/util/event_bus_test.cpp

@@ -0,0 +1,106 @@
+#include "vdm/util/event_bus.hpp"
+
+#include <atomic>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "vtest.hpp"
+
+using vdm::EventBus;
+
+namespace {
+struct Progress {
+    int task;
+    long downloaded;
+};
+struct StateChange {
+    int task;
+    std::string state;
+};
+}  // namespace
+
+VT_TEST(bus_delivers_to_matching_type_only) {
+    EventBus bus;
+    int progress_hits = 0;
+    int state_hits = 0;
+
+    bus.subscribe<Progress>([&](const Progress &p) {
+        ++progress_hits;
+        VT_CHECK_EQ(p.task, 7);
+    });
+    bus.subscribe<StateChange>([&](const StateChange &) { ++state_hits; });
+
+    bus.publish(Progress{7, 1024});
+    VT_CHECK_EQ(progress_hits, 1);
+    VT_CHECK_EQ(state_hits, 0);
+
+    bus.publish(StateChange{7, "downloading"});
+    VT_CHECK_EQ(progress_hits, 1);
+    VT_CHECK_EQ(state_hits, 1);
+}
+
+VT_TEST(bus_invokes_in_registration_order) {
+    EventBus bus;
+    std::vector<int> order;
+    bus.subscribe<Progress>([&](const Progress &) { order.push_back(1); });
+    bus.subscribe<Progress>([&](const Progress &) { order.push_back(2); });
+    bus.subscribe<Progress>([&](const Progress &) { order.push_back(3); });
+    bus.publish(Progress{0, 0});
+    VT_REQUIRE(order.size() == 3);
+    VT_CHECK_EQ(order[0], 1);
+    VT_CHECK_EQ(order[1], 2);
+    VT_CHECK_EQ(order[2], 3);
+}
+
+VT_TEST(bus_unsubscribe_stops_delivery) {
+    EventBus bus;
+    int hits = 0;
+    auto tok = bus.subscribe<Progress>([&](const Progress &) { ++hits; });
+    bus.publish(Progress{0, 0});
+    bus.unsubscribe(tok);
+    bus.publish(Progress{0, 0});
+    VT_CHECK_EQ(hits, 1);
+}
+
+VT_TEST(bus_scoped_subscription_auto_unsubscribes) {
+    EventBus bus;
+    int hits = 0;
+    {
+        auto sub = bus.subscribe_scoped<Progress>([&](const Progress &) { ++hits; });
+        bus.publish(Progress{0, 0});
+    }
+    bus.publish(Progress{0, 0});
+    VT_CHECK_EQ(hits, 1);
+}
+
+VT_TEST(bus_handler_may_unsubscribe_itself_during_dispatch) {
+    EventBus bus;
+    int hits = 0;
+    EventBus::Token tok = EventBus::kInvalid;
+    tok = bus.subscribe<Progress>([&](const Progress &) {
+        ++hits;
+        bus.unsubscribe(tok);  // must not deadlock or invalidate the dispatch loop
+    });
+    bus.publish(Progress{0, 0});
+    bus.publish(Progress{0, 0});
+    VT_CHECK_EQ(hits, 1);
+}
+
+VT_TEST(bus_concurrent_publish_is_safe) {
+    EventBus bus;
+    std::atomic<long> total{0};
+    bus.subscribe<Progress>([&](const Progress &p) { total += p.downloaded; });
+
+    constexpr int kThreads = 8;
+    constexpr int kPerThread = 2000;
+    std::vector<std::jthread> ts;
+    for (int i = 0; i < kThreads; ++i)
+        ts.emplace_back([&, i] {
+            for (int j = 0; j < kPerThread; ++j)
+                bus.publish(Progress{i, 1});
+        });
+    ts.clear();  // join
+
+    VT_CHECK_EQ(total.load(), static_cast<long>(kThreads) * kPerThread);
+}

core/tests/util/log_test.cpp

@@ -0,0 +1,64 @@
+#include "vdm/util/log.hpp"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "vtest.hpp"
+
+using vdm::CallbackSink;
+using vdm::LogLevel;
+using vdm::LogRecord;
+
+namespace {
+struct Captured {
+    LogLevel level;
+    std::string category;
+    std::string message;
+};
+
+// Restores the null sink when it goes out of scope, so tests don't leak a sink.
+struct SinkGuard {
+    ~SinkGuard() { vdm::set_log_sink(nullptr); }
+};
+}  // namespace
+
+VT_TEST(log_discards_when_no_sink) {
+    SinkGuard g;
+    vdm::set_log_sink(nullptr);
+    // Must not crash and must not format: this just has to be a no-op.
+    VDM_LOG_INFO("test", "value {}", 123);
+    VT_CHECK(!vdm::detail::log_wants(LogLevel::error));
+}
+
+VT_TEST(log_forwards_to_sink_with_format) {
+    SinkGuard g;
+    auto hits = std::make_shared<std::vector<Captured>>();
+    vdm::set_log_sink(std::make_shared<CallbackSink>([hits](const LogRecord &r) {
+        hits->push_back({r.level, std::string(r.category), r.message});
+    }));
+
+    VDM_LOG_WARN("probe", "HEAD {} -> {}", "https://x/y", 405);
+    VT_REQUIRE(hits->size() == 1);
+    VT_CHECK_EQ((*hits)[0].level, LogLevel::warn);
+    VT_CHECK_EQ((*hits)[0].category, std::string("probe"));
+    VT_CHECK_EQ((*hits)[0].message, std::string("HEAD https://x/y -> 405"));
+}
+
+VT_TEST(log_level_filter_skips_below_min) {
+    SinkGuard g;
+    auto count = std::make_shared<int>(0);
+    vdm::set_log_sink(std::make_shared<CallbackSink>(
+        [count](const LogRecord &) { ++*count; }, LogLevel::warn));
+
+    VDM_LOG_DEBUG("x", "no");
+    VDM_LOG_INFO("x", "no");
+    VDM_LOG_WARN("x", "yes");
+    VDM_LOG_ERROR("x", "yes");
+    VT_CHECK_EQ(*count, 2);
+}
+
+VT_TEST(log_level_name_is_stable) {
+    VT_CHECK_EQ(vdm::log_level_name(LogLevel::trace), std::string_view("trace"));
+    VT_CHECK_EQ(vdm::log_level_name(LogLevel::error), std::string_view("error"));
+}

core/tests/util/result_test.cpp

@@ -0,0 +1,108 @@
+#include "vdm/util/result.hpp"
+
+#include <string>
+
+#include "vtest.hpp"
+
+using vdm::Err;
+using vdm::Error;
+using vdm::ErrorInfo;
+using vdm::Result;
+
+VT_TEST(result_holds_value) {
+    Result<int> r = 42;
+    VT_REQUIRE(r.has_value());
+    VT_CHECK(static_cast<bool>(r));
+    VT_CHECK_EQ(r.value(), 42);
+    VT_CHECK_EQ(*r, 42);
+    VT_CHECK_EQ(r.code(), Error::ok);
+}
+
+VT_TEST(result_holds_error) {
+    Result<int> r = Err{Error::timeout, "HEAD stalled"};
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.code(), Error::timeout);
+    VT_CHECK_EQ(r.error().code, Error::timeout);
+    VT_CHECK(r.error().retryable);
+    VT_CHECK_EQ(r.value_or(-1), -1);
+}
+
+VT_TEST(result_from_bare_error_code) {
+    Result<std::string> r = Error::not_found;
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::not_found);
+    VT_CHECK_EQ(r.error().http_status, 0);
+}
+
+VT_TEST(result_void_ok) {
+    Result<void> r = vdm::ok();
+    VT_CHECK(r.has_value());
+    VT_CHECK_EQ(r.code(), Error::ok);
+}
+
+VT_TEST(result_void_error) {
+    Result<void> r = Err{Error::disk_full, "temp dir"};
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.code(), Error::disk_full);
+    VT_CHECK(!r.error().retryable);
+}
+
+VT_TEST(result_transform_chains_on_success) {
+    Result<int> r = 21;
+    auto doubled = r.transform([](int v) { return v * 2; });
+    VT_REQUIRE(doubled.has_value());
+    VT_CHECK_EQ(doubled.value(), 42);
+}
+
+VT_TEST(result_and_then_short_circuits_on_error) {
+    Result<int> r = Err{Error::connection_reset, "peer RST"};
+    int calls = 0;
+    auto next = r.and_then([&](int v) -> std::expected<int, ErrorInfo> {
+        ++calls;
+        return v + 1;
+    });
+    VT_CHECK_EQ(calls, 0);
+    VT_REQUIRE(!next.has_value());
+    VT_CHECK_EQ(next.error().code, Error::connection_reset);
+}
+
+namespace {
+Result<int> parse_positive(int n) {
+    if (n < 0)
+        return Err{Error::internal, "negative"};
+    return n;
+}
+
+Result<int> add_two_positives(int a, int b) {
+    VDM_TRY_ASSIGN(auto x, parse_positive(a));
+    VDM_TRY_ASSIGN(auto y, parse_positive(b));
+    return x + y;
+}
+}  // namespace
+
+VT_TEST(vdm_try_assign_propagates) {
+    auto good = add_two_positives(2, 3);
+    VT_REQUIRE(good.has_value());
+    VT_CHECK_EQ(good.value(), 5);
+
+    auto bad = add_two_positives(2, -1);
+    VT_REQUIRE(!bad.has_value());
+    VT_CHECK_EQ(bad.error().code, Error::internal);
+}
+
+VT_TEST(error_info_to_string) {
+    ErrorInfo e{Error::http_server_error, "upstream", 503};
+    VT_CHECK_EQ(e.to_string(), std::string("http_server_error: upstream (HTTP 503)"));
+    VT_CHECK(e.retryable);
+
+    ErrorInfo bare{Error::canceled};
+    VT_CHECK_EQ(bare.to_string(), std::string("canceled"));
+}
+
+VT_TEST(error_name_and_retryable_cover_enum) {
+    VT_CHECK_EQ(vdm::error_name(Error::server_file_changed),
+                std::string_view("server_file_changed"));
+    VT_CHECK(!vdm::is_retryable(Error::server_file_changed));
+    VT_CHECK(!vdm::is_retryable(Error::checksum_mismatch));
+    VT_CHECK(vdm::is_retryable(Error::timeout));
+}

core/tests/util/thread_pool_test.cpp

@@ -0,0 +1,70 @@
+#include "vdm/util/thread_pool.hpp"
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include "vtest.hpp"
+
+using vdm::ThreadPool;
+
+VT_TEST(pool_runs_submitted_task_and_returns_value) {
+    ThreadPool pool(2);
+    auto f = pool.submit([](int a, int b) { return a + b; }, 20, 22);
+    VT_CHECK_EQ(f.get(), 42);
+}
+
+VT_TEST(pool_default_size_is_at_least_one) {
+    ThreadPool pool;
+    VT_CHECK(pool.size() >= 1);
+}
+
+VT_TEST(pool_runs_many_tasks_across_workers) {
+    ThreadPool pool(4);
+    constexpr int kN = 500;
+    std::atomic<int> sum{0};
+    std::vector<std::future<void>> fs;
+    fs.reserve(kN);
+    for (int i = 0; i < kN; ++i)
+        fs.push_back(pool.submit([&sum, i] { sum += i; }));
+    for (auto &f : fs)
+        f.get();
+    VT_CHECK_EQ(sum.load(), kN * (kN - 1) / 2);
+}
+
+VT_TEST(pool_propagates_exceptions_through_future) {
+    ThreadPool pool(1);
+    auto f = pool.submit([]() -> int { throw std::runtime_error("boom"); });
+    bool threw = false;
+    try {
+        f.get();
+    } catch (const std::runtime_error &) {
+        threw = true;
+    }
+    VT_CHECK(threw);
+}
+
+VT_TEST(pool_drains_queued_tasks_on_destruction) {
+    std::atomic<int> done{0};
+    {
+        ThreadPool pool(2);
+        for (int i = 0; i < 50; ++i)
+            pool.submit([&done] {
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+                ++done;
+            });
+        // pool dtor here: must let all 50 finish
+    }
+    VT_CHECK_EQ(done.load(), 50);
+}
+
+VT_TEST(pool_future_from_void_task_is_waitable) {
+    ThreadPool pool(2);
+    std::atomic<bool> ran{false};
+    auto f = pool.submit([&ran] { ran = true; });
+    f.get();
+    VT_CHECK(ran.load());
+}