From ddf36e848aa504348bfd3249c49a4c45cc5bffbf Mon Sep 17 00:00:00 2001 From: sami Date: Wed, 9 Sep 2026 19:03:11 +0400 Subject: [PATCH] =?UTF-8?q?core:=20add=20util=20layer=20=E2=80=94=20Result?= =?UTF-8?q?,=20Error=20taxonomy,=20bytes,=20event=20bus,=20pool,=20log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 over std::expected, Result, 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 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/include/vdm/README.md | 73 +++++++++++++ core/include/vdm/util/.gitkeep | 0 core/include/vdm/util/bytes.hpp | 120 +++++++++++++++++++++ core/include/vdm/util/error.hpp | 107 ++++++++++++++++++ core/include/vdm/util/event_bus.hpp | 150 ++++++++++++++++++++++++++ core/include/vdm/util/log.hpp | 84 +++++++++++++++ core/include/vdm/util/result.hpp | 145 +++++++++++++++++++++++++ core/include/vdm/util/thread_pool.hpp | 82 ++++++++++++++ core/src/util/.gitkeep | 0 core/src/util/error.cpp | 96 +++++++++++++++++ core/src/util/log.cpp | 53 +++++++++ core/src/util/thread_pool.cpp | 49 +++++++++ core/tests/util/bytes_test.cpp | 94 ++++++++++++++++ core/tests/util/event_bus_test.cpp | 106 ++++++++++++++++++ core/tests/util/log_test.cpp | 64 +++++++++++ core/tests/util/result_test.cpp | 108 +++++++++++++++++++ core/tests/util/thread_pool_test.cpp | 70 ++++++++++++ 17 files changed, 1401 insertions(+) create mode 100644 core/include/vdm/README.md delete mode 100644 core/include/vdm/util/.gitkeep create mode 100644 core/include/vdm/util/bytes.hpp create mode 100644 core/include/vdm/util/error.hpp create mode 100644 core/include/vdm/util/event_bus.hpp create mode 100644 core/include/vdm/util/log.hpp create mode 100644 core/include/vdm/util/result.hpp create mode 100644 core/include/vdm/util/thread_pool.hpp delete mode 100644 core/src/util/.gitkeep create mode 100644 core/src/util/error.cpp create mode 100644 core/src/util/log.cpp create mode 100644 core/src/util/thread_pool.cpp create mode 100644 core/tests/util/bytes_test.cpp create mode 100644 core/tests/util/event_bus_test.cpp create mode 100644 core/tests/util/log_test.cpp create mode 100644 core/tests/util/result_test.cpp create mode 100644 core/tests/util/thread_pool_test.cpp diff --git a/core/include/vdm/README.md b/core/include/vdm/README.md new file mode 100644 index 0000000..7bbdfda --- /dev/null +++ b/core/include/vdm/README.md @@ -0,0 +1,73 @@ +# `libveloxcore` — public API + +**Status: M1 in progress.** Only `util/` is landed. The download-facing API +(`DownloadSpec`, `DownloadTask`, probe, typed callbacks) arrives with later stages and +is reviewed by DAEMON before M2 (AGENT-CORE DoD). + +Layering (CLAUDE.md §3): this library knows nothing about JSON, SQL, Qt, or RPC. Input is +a spec value; output is bytes on disk plus typed callbacks. DAEMON projects engine state +onto the wire contract's `TaskSummary` / `TaskDetail` / events — see +`core/docs/proto-requests-m1.md` for the shapes that projection needs frozen. + +Every header under `core/include/vdm/` compiles standalone (`-Wall -Wextra -Wpedantic +-Werror`, C++23). Clean under ASan/UBSan and TSan. + +--- + +## `util/` — foundations + +### `vdm/util/error.hpp` + +`enum class Error` — the engine-wide failure taxonomy (network / HTTP / content / local +I/O / metadata / probe / retry / internal). This is CORE's own vocabulary; it is **not** +a wire type. `error_name(Error)` gives a stable snake_case string; `is_retryable(Error)` +is the advisory retry hint the task policy consults. + +`struct ErrorInfo { Error code; std::string context; int http_status; bool retryable; +Error cause; }` — the payload carried by every failed `Result`. `.to_string()` renders +`": (HTTP )"`. + +### `vdm/util/result.hpp` + +`Result` — return-based error channel, a thin wrapper over +`std::expected`. Errors are **returned, never thrown**, on anything that +runs during a transfer. + +- `Result r = 42;` / `Result r = Err{Error::timeout, "..."};` / + `Result r = Error::not_found;` +- `r.has_value()`, `explicit operator bool`, `r.value()` / `*r` / `r->`, `r.error()`, + `r.code()`, `r.value_or(x)` +- monadic `and_then` / `transform` / `transform_error` (forward to `std::expected`) +- `Result` specialization; `vdm::ok()` success sentinel +- `VDM_TRY(expr)` — return the error if `expr` failed +- `VDM_TRY_ASSIGN(auto x, expr)` — bind the value or return the error + +### `vdm/util/bytes.hpp` + +`Byte` / `ByteSpan` / `ConstByteSpan` aliases; `as_bytes(string_view)` / +`as_chars(span)`. Little-endian fixed-width codec `load_le` / `store_le` and a +bounds-checked sequential `ByteReader` (`.u8/.u16/.u32/.u64`, `.raw(n)`, `.lp_string()`, +`.overran()`). Built for the `.veloxpart.meta` reader and the 4-byte NM framing; every +read is bounds-checked and latches on overrun (reader-first, fuzz-ready). + +### `vdm/util/event_bus.hpp` + +`EventBus` — typed, thread-safe in-process pub/sub. `subscribe(fn) -> Token`, +`publish(ev)` (synchronous, calling thread, registration order), `unsubscribe(Token)`, +and RAII `subscribe_scoped` returning a `Subscription`. Handlers may (un)subscribe or +publish during dispatch. Handlers must not throw. Not a hot-path structure — progress is +coalesced to ≤4 Hz upstream. + +### `vdm/util/thread_pool.hpp` + +`ThreadPool` — fixed-size `std::jthread` pool for bounded off-loop work (hashing, fsync +batches, DNS pre-resolve). `submit(fn, args...) -> std::future`; propagates exceptions +through the future; drains already-queued tasks on destruction. **Not** the transfer +loop — `net/` will own one `curl_multi` per dedicated worker. + +### `vdm/util/log.hpp` + +Sink interface — core does no I/O itself. `LogSink` abstract base; DAEMON installs one +via `set_log_sink()`, default discards. `CallbackSink` adapter (with a min-level filter). +`VDM_LOG_{TRACE,DEBUG,INFO,WARN,ERROR}(category, fmt, args...)` — `std::format` syntax, +only formatted when a sink is installed and wants the level. diff --git a/core/include/vdm/util/.gitkeep b/core/include/vdm/util/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/core/include/vdm/util/bytes.hpp b/core/include/vdm/util/bytes.hpp new file mode 100644 index 0000000..5cdb7f5 --- /dev/null +++ b/core/include/vdm/util/bytes.hpp @@ -0,0 +1,120 @@ +// vdm/util/bytes.hpp — byte-span aliases and a little-endian integer codec. +// +// Used by the 4-byte native-messaging framing, the .veloxpart.meta record (docs/04 §5, +// "Little-endian, versioned"), and the write path. Pure, header-only, no allocation. +// +// This header compiles standalone. + +#ifndef VDM_UTIL_BYTES_HPP +#define VDM_UTIL_BYTES_HPP + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static_assert(CHAR_BIT == 8, "vdm assumes 8-bit bytes"); + +namespace vdm { + +using Byte = std::byte; +using ByteSpan = std::span; +using ConstByteSpan = std::span; + +// View the bytes of a string / string_view without copying. +[[nodiscard]] inline ConstByteSpan as_bytes(std::string_view s) noexcept { + return {reinterpret_cast(s.data()), s.size()}; +} + +[[nodiscard]] inline std::string_view as_chars(ConstByteSpan b) noexcept { + return {reinterpret_cast(b.data()), b.size()}; +} + +// --- little-endian fixed-width integer codec ------------------------------------------- +// +// load_le(span) / store_le(span, value): read/write a T from/to the first +// sizeof(T) bytes of the span. The span MUST be large enough — that is a caller +// precondition, asserted in debug builds only. These are not the bounds-checking layer: +// on the .veloxpart.meta parse path (attacker-adjacent, AGENT-CORE §5) callers must +// size-check first and return Error::meta_corrupt, or go through ByteReader below, which +// does the checking. Do not treat the assert as input validation. + +template +[[nodiscard]] T load_le(ConstByteSpan src) noexcept { + assert(src.size() >= sizeof(T) && "load_le: source span too small (caller must size-check)"); + T host{}; + std::memcpy(&host, src.data(), sizeof(T)); + if constexpr (std::endian::native == std::endian::big) + host = std::byteswap(host); + return host; +} + +template +void store_le(ByteSpan dst, T value) noexcept { + assert(dst.size() >= sizeof(T) && "store_le: destination span too small"); + if constexpr (std::endian::native == std::endian::big) + value = std::byteswap(value); + std::memcpy(dst.data(), &value, sizeof(T)); +} + +// Cursor over a byte span for sequential LE reads. Every read is bounds-checked; once a +// read runs past the end the cursor latches into an error state and all further reads +// return zero. Built for the .veloxpart.meta reader ("Write the reader first and fuzz +// it" — AGENT-CORE §5). +class ByteReader { + ConstByteSpan buf_; + std::size_t pos_ = 0; + bool overran_ = false; + + public: + explicit ByteReader(ConstByteSpan buf) noexcept : buf_(buf) {} + + [[nodiscard]] bool overran() const noexcept { return overran_; } + [[nodiscard]] std::size_t offset() const noexcept { return pos_; } + [[nodiscard]] std::size_t remaining() const noexcept { + return overran_ ? 0 : buf_.size() - pos_; + } + + template + [[nodiscard]] T u() noexcept { + if (overran_ || buf_.size() - pos_ < sizeof(T)) { + overran_ = true; + return 0; + } + T v = load_le(buf_.subspan(pos_)); + pos_ += sizeof(T); + return v; + } + + [[nodiscard]] std::uint8_t u8() noexcept { return u(); } + [[nodiscard]] std::uint16_t u16() noexcept { return u(); } + [[nodiscard]] std::uint32_t u32() noexcept { return u(); } + [[nodiscard]] std::uint64_t u64() noexcept { return u(); } + + // Read `n` raw bytes; empty span (and overran latched) if short. + [[nodiscard]] ConstByteSpan raw(std::size_t n) noexcept { + if (overran_ || buf_.size() - pos_ < n) { + overran_ = true; + return {}; + } + ConstByteSpan out = buf_.subspan(pos_, n); + pos_ += n; + return out; + } + + // Length-prefixed UTF-8 blob: u32 length then that many bytes (docs/04 §5 + // "length-prefixed UTF-8"). Returns empty view on overrun. + [[nodiscard]] std::string_view lp_string() noexcept { + std::uint32_t n = u32(); + return as_chars(raw(n)); + } +}; + +} // namespace vdm + +#endif // VDM_UTIL_BYTES_HPP diff --git a/core/include/vdm/util/error.hpp b/core/include/vdm/util/error.hpp new file mode 100644 index 0000000..98efde5 --- /dev/null +++ b/core/include/vdm/util/error.hpp @@ -0,0 +1,107 @@ +// vdm/util/error.hpp — the engine-wide failure taxonomy. +// +// core/ never imports a protocol header (see CLAUDE.md layering rule), so this enum is +// CORE's own vocabulary. DAEMON projects it onto the wire `TaskSummary.error.code`. +// Keep it lossless-mappable: see core/docs/proto-requests-m1.md (B1). +// +// This header compiles standalone. + +#ifndef VDM_UTIL_ERROR_HPP +#define VDM_UTIL_ERROR_HPP + +#include +#include +#include + +namespace vdm { + +// Ordered by rough class: cancellation, network, HTTP, content, local I/O, metadata, +// probe, retry, internal. New values append within their class; never renumber — DAEMON +// and the meta-file writer key off the numeric value. +enum class Error : std::uint16_t { + ok = 0, + + // --- cancellation --- + canceled = 100, + + // --- network / transport (mostly retryable) --- + resolve_failed = 200, + connect_failed, + tls_failed, + connection_reset, + timeout, + too_many_redirects, + + // --- HTTP status (carry http_status) --- + http_client_error = 300, // 4xx, unclassified + http_server_error, // 5xx + auth_required, // 401 / 407 — task pauses, not terminal + forbidden, // 403 after the referrer retry + not_found, // 404 + range_not_satisfiable, // 416 — re-probe / re-split + gone, // 410 / expired signed URL, no mirror + + // --- content / semantics --- + server_file_changed = 400, // 200 where 206 expected, or If-Range / ETag drift + content_length_mismatch, + checksum_mismatch, + + // --- local I/O --- + disk_full = 500, + io_error, + path_rejected, // outside allowed roots or not writable + permission_denied, // EACCES on the destination + + // --- resume metadata --- + meta_corrupt = 600, // bad magic / failed CRC + meta_version_unsupported, // written by a newer engine + + // --- probe --- + probe_failed = 700, + unsupported_url_scheme, + + // --- retry --- + max_retries_exhausted = 800, // `cause` names the last underlying Error + + // --- catch-all --- + internal = 900, +}; + +// Stable snake_case name, matching the table in proto-requests-m1.md. Never returns +// nullptr; unknown values yield "unknown". +[[nodiscard]] std::string_view error_name(Error e) noexcept; + +// True for errors the task state machine may retry with backoff. Advisory: the task +// policy still owns the final decision (max_retries, mirror availability). +[[nodiscard]] bool is_retryable(Error e) noexcept; + +// Carried alongside every failed Result. Construction is off the transfer hot path +// (the curl write callback only ever returns success or a bare io_error), so the +// std::string here is fine. +struct ErrorInfo { + Error code = Error::internal; + std::string context; // human-readable, for logs and event.notify bodies + int http_status = 0; // 0 when not HTTP-derived + bool retryable = false; // snapshot of is_retryable(code) at construction, may be + // overridden by the caller (e.g. probe_failed) + Error cause = Error::ok; // underlying error when `code` is a wrapper + // (max_retries_exhausted) + + ErrorInfo() = default; + + explicit ErrorInfo(Error c, std::string ctx = {}, int status = 0) + : code(c), + context(std::move(ctx)), + http_status(status), + retryable(is_retryable(c)) {} + + [[nodiscard]] std::string_view name() const noexcept { return error_name(code); } + + // ": " (or just "" when context is empty), plus " (HTTP )" + // when http_status is set. + [[nodiscard]] std::string to_string() const; +}; + +} // namespace vdm + +#endif // VDM_UTIL_ERROR_HPP diff --git a/core/include/vdm/util/event_bus.hpp b/core/include/vdm/util/event_bus.hpp new file mode 100644 index 0000000..b3d0c48 --- /dev/null +++ b/core/include/vdm/util/event_bus.hpp @@ -0,0 +1,150 @@ +// vdm/util/event_bus.hpp — a typed, thread-safe in-process publish/subscribe bus. +// +// The engine emits typed callbacks (AGENT-CORE: "public API ... emits typed callbacks"). +// Internally, subsystems fan progress / state / probe results out through one bus so the +// task layer and the (daemon-injected) reporting shim don't wire up N direct callbacks. +// +// Contract: +// - subscribe(fn) registers fn for events of exactly type E; returns a token. +// - publish(ev) invokes every handler registered for E, on the calling thread, +// synchronously, in registration order. +// - unsubscribe(token) removes a handler from the registry. Safe to call from inside a +// handler and from another thread. +// - Handlers must not throw. A throwing handler calls std::terminate (they run on the +// transfer threads; there is no sensible recovery and swallowing hides bugs). +// +// unsubscribe is NOT a quiesce point. When it returns (and likewise when ~Subscription +// returns), a handler for that token may still be running on another thread — publish() +// snapshots the handler list and then invokes unlocked, so a dispatch already in flight +// keeps going against a copy. A handler that captured `this` can therefore fire after the +// owner began tearing down. Callers must not use unsubscribe for lifetime safety. +// +// download_task (stage 8) subscribes handlers that close over task state, so it needs a +// real drain/quiesce (stop accepting the event, then wait for in-flight dispatch to +// finish) built at that layer before it relies on teardown ordering. This bus does not +// provide one and should not grow one just for that caller. +// +// Not a hot-path structure: progress events are coalesced to <=4 Hz upstream (see the +// wire contract's event.task.progress). A shared_mutex + copy-on-dispatch is the right +// weight here; a lock-free SPSC ring is a later optimisation if a profile asks for it. +// +// This header compiles standalone. + +#ifndef VDM_UTIL_EVENT_BUS_HPP +#define VDM_UTIL_EVENT_BUS_HPP + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace vdm { + +class EventBus { + public: + using Token = std::uint64_t; + static constexpr Token kInvalid = 0; + + EventBus() = default; + EventBus(const EventBus &) = delete; + EventBus &operator=(const EventBus &) = delete; + + template + Token subscribe(std::function handler) { + std::unique_lock lk(mu_); + Token tok = ++last_token_; + auto &slot = channels_[std::type_index(typeid(E))]; + slot.push_back(Entry{ + tok, + [h = std::move(handler)](const void *ev) { + h(*static_cast(ev)); + }, + }); + return tok; + } + + template + void publish(const E &ev) const { + // Copy the handler list under a shared lock, then invoke unlocked so a handler + // may (un)subscribe or publish without deadlocking or iterator-invalidating. + std::vector snapshot; + { + std::shared_lock lk(mu_); + auto it = channels_.find(std::type_index(typeid(E))); + if (it == channels_.end()) + return; + snapshot = it->second; + } + for (const auto &e : snapshot) + e.fn(&ev); + } + + void unsubscribe(Token tok) { + if (tok == kInvalid) + return; + std::unique_lock lk(mu_); + for (auto &[_, entries] : channels_) { + auto it = std::find_if(entries.begin(), entries.end(), + [tok](const Entry &e) { return e.token == tok; }); + if (it != entries.end()) { + entries.erase(it); + return; + } + } + } + + // RAII holder: unsubscribes on destruction. Non-copyable, movable. + class Subscription { + EventBus *bus_ = nullptr; + Token tok_ = kInvalid; + + public: + Subscription() = default; + Subscription(EventBus *bus, Token tok) : bus_(bus), tok_(tok) {} + Subscription(Subscription &&o) noexcept + : bus_(std::exchange(o.bus_, nullptr)), tok_(std::exchange(o.tok_, kInvalid)) {} + Subscription &operator=(Subscription &&o) noexcept { + if (this != &o) { + reset(); + bus_ = std::exchange(o.bus_, nullptr); + tok_ = std::exchange(o.tok_, kInvalid); + } + return *this; + } + Subscription(const Subscription &) = delete; + Subscription &operator=(const Subscription &) = delete; + ~Subscription() { reset(); } + + void reset() { + if (bus_ && tok_ != kInvalid) + bus_->unsubscribe(tok_); + bus_ = nullptr; + tok_ = kInvalid; + } + [[nodiscard]] Token token() const noexcept { return tok_; } + }; + + template + Subscription subscribe_scoped(std::function handler) { + return Subscription(this, subscribe(std::move(handler))); + } + + private: + struct Entry { + Token token; + std::function fn; + }; + + mutable std::shared_mutex mu_; + std::unordered_map> channels_; + Token last_token_ = kInvalid; +}; + +} // namespace vdm + +#endif // VDM_UTIL_EVENT_BUS_HPP diff --git a/core/include/vdm/util/log.hpp b/core/include/vdm/util/log.hpp new file mode 100644 index 0000000..0a44bb6 --- /dev/null +++ b/core/include/vdm/util/log.hpp @@ -0,0 +1,84 @@ +// vdm/util/log.hpp — a sink interface the daemon fills in; core does no I/O itself. +// +// core/ must not open files or write to stderr (docs/01 §3: "Input: a DownloadSpec. +// Output: bytes on disk + callbacks."). So logging is an interface: DAEMON installs a +// sink that forwards into velox.log; until then the default sink discards everything. +// +// Usage: +// VDM_LOG_INFO("probe", "HEAD {} -> {}", url, status); +// The format string is std::format syntax and is only formatted if a sink is installed +// and wants that level. +// +// This header compiles standalone. + +#ifndef VDM_UTIL_LOG_HPP +#define VDM_UTIL_LOG_HPP + +#include +#include +#include +#include +#include +#include + +namespace vdm { + +enum class LogLevel { trace, debug, info, warn, error }; + +[[nodiscard]] std::string_view log_level_name(LogLevel) noexcept; + +struct LogRecord { + LogLevel level; + std::string_view category; // short static tag: "probe", "segmenter", "io", ... + std::string message; +}; + +class LogSink { + public: + virtual ~LogSink() = default; + virtual void write(const LogRecord &) = 0; + // Fast pre-check so callers can skip formatting entirely. Default: everything. + [[nodiscard]] virtual bool enabled(LogLevel) const { return true; } +}; + +// Adapter: wrap a plain callable as a sink (handy for tests and simple daemon glue). +class CallbackSink final : public LogSink { + public: + using Fn = std::function; + explicit CallbackSink(Fn fn, LogLevel min = LogLevel::trace) + : fn_(std::move(fn)), min_(min) {} + void write(const LogRecord &r) override { fn_(r); } + [[nodiscard]] bool enabled(LogLevel l) const override { return l >= min_; } + + private: + Fn fn_; + LogLevel min_; +}; + +// Process-wide sink. Thread-safe. Passing nullptr restores the discarding sink. +void set_log_sink(std::shared_ptr); +[[nodiscard]] std::shared_ptr log_sink(); + +// Non-macro entry point (macros below call this). +void log_emit(LogLevel, std::string_view category, std::string message); + +namespace detail { +[[nodiscard]] bool log_wants(LogLevel); // sink installed && sink.enabled(level) +} + +} // namespace vdm + +#define VDM_LOG(level, category, ...) \ + do { \ + if (::vdm::detail::log_wants(level)) \ + ::vdm::log_emit((level), (category), \ + std::format(__VA_ARGS__)); \ + } while (0) + +#define VDM_LOG_TRACE(cat, ...) VDM_LOG(::vdm::LogLevel::trace, cat, __VA_ARGS__) +#define VDM_LOG_DEBUG(cat, ...) VDM_LOG(::vdm::LogLevel::debug, cat, __VA_ARGS__) +#define VDM_LOG_INFO(cat, ...) VDM_LOG(::vdm::LogLevel::info, cat, __VA_ARGS__) +#define VDM_LOG_WARN(cat, ...) VDM_LOG(::vdm::LogLevel::warn, cat, __VA_ARGS__) +#define VDM_LOG_ERROR(cat, ...) VDM_LOG(::vdm::LogLevel::error, cat, __VA_ARGS__) + +#endif // VDM_UTIL_LOG_HPP diff --git a/core/include/vdm/util/result.hpp b/core/include/vdm/util/result.hpp new file mode 100644 index 0000000..2087bda --- /dev/null +++ b/core/include/vdm/util/result.hpp @@ -0,0 +1,145 @@ +// vdm/util/result.hpp — the return-based error channel for the transfer path. +// +// Errors are returned, never thrown, on anything that runs during a download +// (AGENT-CORE brief). Result is a thin wrapper over std::expected +// so we keep a stable spelling and a few helpers; the monadic surface forwards to +// std::expected. +// +// This header compiles standalone. + +#ifndef VDM_UTIL_RESULT_HPP +#define VDM_UTIL_RESULT_HPP + +#include +#include +#include +#include + +#include "vdm/util/error.hpp" + +namespace vdm { + +// Alias for the "failure" side, so call sites read `return Err{Error::timeout, "..."}`. +using Err = ErrorInfo; + +template +class [[nodiscard]] Result { + using storage = std::expected; + storage exp_; + + public: + using value_type = T; + using error_type = ErrorInfo; + + // Success construction: implicit from a T (or anything convertible to T). + template && + !std::is_same_v, Result>>> + Result(U &&value) : exp_(std::in_place, std::forward(value)) {} + + Result() requires std::is_default_constructible_v : exp_(std::in_place) {} + + // Failure construction: implicit from an ErrorInfo/Err. + Result(ErrorInfo error) : exp_(std::unexpected(std::move(error))) {} + Result(Error code) : exp_(std::unexpected(ErrorInfo(code))) {} + + [[nodiscard]] bool has_value() const noexcept { return exp_.has_value(); } + explicit operator bool() const noexcept { return exp_.has_value(); } + + T &value() & { return exp_.value(); } + const T &value() const & { return exp_.value(); } + T &&value() && { return std::move(exp_).value(); } + + T *operator->() noexcept { return &*exp_; } + const T *operator->() const noexcept { return &*exp_; } + T &operator*() & noexcept { return *exp_; } + const T &operator*() const & noexcept { return *exp_; } + T &&operator*() && noexcept { return *std::move(exp_); } + + const ErrorInfo &error() const & { return exp_.error(); } + ErrorInfo &&error() && { return std::move(exp_).error(); } + + [[nodiscard]] Error code() const noexcept { + return exp_.has_value() ? Error::ok : exp_.error().code; + } + + template + T value_or(U &&fallback) const & { + return exp_.value_or(std::forward(fallback)); + } + + // Monadic forwarding — see std::expected. `and_then` chains Result-returning + // callables; `transform` maps the value; `transform_error` rewrites the failure. + template + auto and_then(F &&f) & { return exp_.and_then(std::forward(f)); } + template + auto and_then(F &&f) const & { return exp_.and_then(std::forward(f)); } + template + auto and_then(F &&f) && { return std::move(exp_).and_then(std::forward(f)); } + + template + auto transform(F &&f) & { return exp_.transform(std::forward(f)); } + template + auto transform(F &&f) const & { return exp_.transform(std::forward(f)); } + template + auto transform(F &&f) && { return std::move(exp_).transform(std::forward(f)); } + + template + auto transform_error(F &&f) const & { + return exp_.transform_error(std::forward(f)); + } +}; + +// Result — success carries nothing; failure still carries ErrorInfo. +template <> +class [[nodiscard]] Result { + std::expected exp_; + + public: + using value_type = void; + using error_type = ErrorInfo; + + Result() : exp_() {} + Result(ErrorInfo error) : exp_(std::unexpected(std::move(error))) {} + Result(Error code) : exp_(std::unexpected(ErrorInfo(code))) {} + + [[nodiscard]] bool has_value() const noexcept { return exp_.has_value(); } + explicit operator bool() const noexcept { return exp_.has_value(); } + void value() const { exp_.value(); } + + const ErrorInfo &error() const & { return exp_.error(); } + ErrorInfo &&error() && { return std::move(exp_).error(); } + + [[nodiscard]] Error code() const noexcept { + return exp_.has_value() ? Error::ok : exp_.error().code; + } +}; + +// Explicit success sentinel for Result returns that reads better than `return {}`. +inline Result ok() { return {}; } + +} // namespace vdm + +#define VDM_DETAIL_CAT_(a, b) a##b +#define VDM_DETAIL_CAT(a, b) VDM_DETAIL_CAT_(a, b) + +// VDM_TRY(expr): evaluate a Result-returning expression; on failure, return its error +// from the enclosing function (which must itself return a Result<...>). On success the +// macro yields nothing — use VDM_TRY_ASSIGN to bind the value. +#define VDM_TRY(expr) \ + do { \ + auto _vdm_r = (expr); \ + if (!_vdm_r.has_value()) \ + return ::vdm::ErrorInfo(std::move(_vdm_r).error()); \ + } while (0) + +// VDM_TRY_ASSIGN(decl, expr): bind `decl` to the value of a successful Result, else +// return its error. Usage: VDM_TRY_ASSIGN(auto n, read_some()); +#define VDM_TRY_ASSIGN(decl, expr) \ + auto VDM_DETAIL_CAT(_vdm_tmp_, __LINE__) = (expr); \ + if (!VDM_DETAIL_CAT(_vdm_tmp_, __LINE__).has_value()) \ + return ::vdm::ErrorInfo( \ + std::move(VDM_DETAIL_CAT(_vdm_tmp_, __LINE__)).error()); \ + decl = *std::move(VDM_DETAIL_CAT(_vdm_tmp_, __LINE__)) + +#endif // VDM_UTIL_RESULT_HPP diff --git a/core/include/vdm/util/thread_pool.hpp b/core/include/vdm/util/thread_pool.hpp new file mode 100644 index 0000000..948b86f --- /dev/null +++ b/core/include/vdm/util/thread_pool.hpp @@ -0,0 +1,82 @@ +// vdm/util/thread_pool.hpp — a fixed-size worker pool for off-loop utility work. +// +// AGENT-CORE bans naked pthread; workers are std::jthread. This pool is for bounded +// utility tasks (hashing a completed file, fsync batches, DNS pre-resolve). The curl +// transfer loop is NOT built on this — net/ owns one curl_multi per dedicated worker +// thread (docs/01 §2). Keep those concerns separate. +// +// Shutdown is drain-only: ~ThreadPool lets every already-queued job run, so teardown is +// bounded by the slowest queued job. That is fine for the short tasks above. DAEMON's +// graceful-shutdown path will need a cancel-instead-of-drain mode (stop, discard the +// pending queue, join) — add it when something actually queues long-running work, not +// before. +// +// This header compiles standalone. + +#ifndef VDM_UTIL_THREAD_POOL_HPP +#define VDM_UTIL_THREAD_POOL_HPP + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace vdm { + +class ThreadPool { + public: + // Defaults to hardware_concurrency() (min 1). Pass an explicit count for tests. + explicit ThreadPool(std::size_t threads = 0); + + // Stops accepting work, wakes all workers, lets in-flight and already-queued tasks + // drain, then joins. A task still queued at shutdown DOES run. + ~ThreadPool(); + + ThreadPool(const ThreadPool &) = delete; + ThreadPool &operator=(const ThreadPool &) = delete; + + [[nodiscard]] std::size_t size() const noexcept { return workers_.size(); } + + // Enqueue `fn(args...)`; returns a future for its result. Throws std::runtime_error + // if the pool is already shutting down. + template + auto submit(F &&fn, Args &&...args) + -> std::future> { + using R = std::invoke_result_t; + + auto task = std::make_shared>( + [f = std::forward(fn), + ... a = std::forward(args)]() mutable -> R { + return std::invoke(std::move(f), std::move(a)...); + }); + std::future fut = task->get_future(); + + { + std::lock_guard lk(mu_); + if (stopping_) + throw std::runtime_error("ThreadPool::submit after shutdown"); + jobs_.emplace([task = std::move(task)] { (*task)(); }); + } + cv_.notify_one(); + return fut; + } + + private: + void worker_loop(); + + std::vector workers_; + std::queue> jobs_; + std::mutex mu_; + std::condition_variable cv_; + bool stopping_ = false; +}; + +} // namespace vdm + +#endif // VDM_UTIL_THREAD_POOL_HPP diff --git a/core/src/util/.gitkeep b/core/src/util/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/core/src/util/error.cpp b/core/src/util/error.cpp new file mode 100644 index 0000000..e0afb72 --- /dev/null +++ b/core/src/util/error.cpp @@ -0,0 +1,96 @@ +// vdm/util/error.cpp + +#include "vdm/util/error.hpp" + +namespace vdm { + +std::string_view error_name(Error e) noexcept { + switch (e) { + case Error::ok: return "ok"; + case Error::canceled: return "canceled"; + case Error::resolve_failed: return "resolve_failed"; + case Error::connect_failed: return "connect_failed"; + case Error::tls_failed: return "tls_failed"; + case Error::connection_reset: return "connection_reset"; + case Error::timeout: return "timeout"; + case Error::too_many_redirects: return "too_many_redirects"; + case Error::http_client_error: return "http_client_error"; + case Error::http_server_error: return "http_server_error"; + case Error::auth_required: return "auth_required"; + case Error::forbidden: return "forbidden"; + case Error::not_found: return "not_found"; + case Error::range_not_satisfiable: return "range_not_satisfiable"; + case Error::gone: return "gone"; + case Error::server_file_changed: return "server_file_changed"; + case Error::content_length_mismatch: return "content_length_mismatch"; + case Error::checksum_mismatch: return "checksum_mismatch"; + case Error::disk_full: return "disk_full"; + case Error::io_error: return "io_error"; + case Error::path_rejected: return "path_rejected"; + case Error::permission_denied: return "permission_denied"; + case Error::meta_corrupt: return "meta_corrupt"; + case Error::meta_version_unsupported: return "meta_version_unsupported"; + case Error::probe_failed: return "probe_failed"; + case Error::unsupported_url_scheme: return "unsupported_url_scheme"; + case Error::max_retries_exhausted: return "max_retries_exhausted"; + case Error::internal: return "internal"; + } + return "unknown"; +} + +bool is_retryable(Error e) noexcept { + // Every value is listed on purpose: no `default:`, so adding an Error without a + // -Werror -Wswitch diagnostic is impossible and the retry decision is forced. + switch (e) { + // retryable + case Error::resolve_failed: + case Error::connect_failed: + case Error::connection_reset: + case Error::timeout: + case Error::http_server_error: + case Error::range_not_satisfiable: // stale metadata → re-probe then retry + case Error::content_length_mismatch: + return true; + + // not retryable + case Error::ok: + case Error::canceled: + case Error::tls_failed: + case Error::too_many_redirects: + case Error::http_client_error: + case Error::auth_required: // resolved by credentials, not a retry + case Error::forbidden: + case Error::not_found: + case Error::gone: + case Error::server_file_changed: // needs a user decision + case Error::checksum_mismatch: + case Error::disk_full: + case Error::io_error: + case Error::path_rejected: + case Error::permission_denied: + case Error::meta_corrupt: + case Error::meta_version_unsupported: + case Error::probe_failed: + case Error::unsupported_url_scheme: + case Error::max_retries_exhausted: + case Error::internal: + return false; + } + return false; // unreachable; silences -Wreturn-type for an out-of-range value +} + +std::string ErrorInfo::to_string() const { + std::string out(name()); + if (!context.empty()) { + out += ": "; + out += context; + } + if (http_status != 0) { + out += " (HTTP "; + out += std::to_string(http_status); + out += ')'; + } + return out; +} + +} // namespace vdm diff --git a/core/src/util/log.cpp b/core/src/util/log.cpp new file mode 100644 index 0000000..59e51d9 --- /dev/null +++ b/core/src/util/log.cpp @@ -0,0 +1,53 @@ +// vdm/util/log.cpp + +#include "vdm/util/log.hpp" + +#include +#include + +namespace vdm { +namespace { + +// Guards assignment; reads take a copy of the shared_ptr under the same lock so a +// concurrent set_log_sink() can't free the sink mid-write. +std::mutex g_mu; +std::shared_ptr g_sink; + +} // namespace + +std::string_view log_level_name(LogLevel l) noexcept { + switch (l) { + case LogLevel::trace: return "trace"; + case LogLevel::debug: return "debug"; + case LogLevel::info: return "info"; + case LogLevel::warn: return "warn"; + case LogLevel::error: return "error"; + } + return "?"; +} + +void set_log_sink(std::shared_ptr sink) { + std::lock_guard lk(g_mu); + g_sink = std::move(sink); +} + +std::shared_ptr log_sink() { + std::lock_guard lk(g_mu); + return g_sink; +} + +namespace detail { +bool log_wants(LogLevel level) { + auto s = log_sink(); + return s && s->enabled(level); +} +} // namespace detail + +void log_emit(LogLevel level, std::string_view category, std::string message) { + auto s = log_sink(); + if (!s || !s->enabled(level)) + return; + s->write(LogRecord{level, category, std::move(message)}); +} + +} // namespace vdm diff --git a/core/src/util/thread_pool.cpp b/core/src/util/thread_pool.cpp new file mode 100644 index 0000000..0d7fd9b --- /dev/null +++ b/core/src/util/thread_pool.cpp @@ -0,0 +1,49 @@ +// vdm/util/thread_pool.cpp + +#include "vdm/util/thread_pool.hpp" + +namespace vdm { + +ThreadPool::ThreadPool(std::size_t threads) { + if (threads == 0) { + threads = std::thread::hardware_concurrency(); + if (threads == 0) + threads = 1; + } + workers_.reserve(threads); + for (std::size_t i = 0; i < threads; ++i) + workers_.emplace_back([this] { worker_loop(); }); +} + +ThreadPool::~ThreadPool() { + { + std::lock_guard lk(mu_); + stopping_ = true; + } + cv_.notify_all(); + // Join here, in the destructor body, while mu_/cv_/jobs_ are still alive. Relying on + // std::jthread's implicit join would run it during member destruction — after cv_ and + // mu_ are already gone, which the workers are still touching. + for (auto &w : workers_) + w.join(); + workers_.clear(); +} + +void ThreadPool::worker_loop() { + for (;;) { + std::function job; + { + std::unique_lock lk(mu_); + cv_.wait(lk, [this] { return stopping_ || !jobs_.empty(); }); + if (jobs_.empty()) { + // Only reached when stopping_ and nothing left to do. + return; + } + job = std::move(jobs_.front()); + jobs_.pop(); + } + job(); + } +} + +} // namespace vdm diff --git a/core/tests/util/bytes_test.cpp b/core/tests/util/bytes_test.cpp new file mode 100644 index 0000000..d5671f3 --- /dev/null +++ b/core/tests/util/bytes_test.cpp @@ -0,0 +1,94 @@ +#include "vdm/util/bytes.hpp" + +#include +#include +#include +#include +#include + +#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 buf{}; + store_le(buf, 0x01020304u); + // little-endian: least significant byte first + VT_CHECK_EQ(std::to_integer(buf[0]), 0x04); + VT_CHECK_EQ(std::to_integer(buf[1]), 0x03); + VT_CHECK_EQ(std::to_integer(buf[2]), 0x02); + VT_CHECK_EQ(std::to_integer(buf[3]), 0x01); + VT_CHECK_EQ(load_le(buf), 0x01020304u); +} + +VT_TEST(le_roundtrip_u64) { + std::array buf{}; + const std::uint64_t v = 0xDEADBEEF0BADF00Dull; + store_le(buf, v); + VT_CHECK_EQ(load_le(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 buf{}; + ByteSpan w(buf); + std::memcpy(buf.data(), "VDMP", 4); + store_le(w.subspan(4), 2); + store_le(w.subspan(6), 0); + store_le(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 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 buf{}; + ByteSpan w(buf); + store_le(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 buf{}; + ByteSpan w(buf); + store_le(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()); +} diff --git a/core/tests/util/event_bus_test.cpp b/core/tests/util/event_bus_test.cpp new file mode 100644 index 0000000..a284b85 --- /dev/null +++ b/core/tests/util/event_bus_test.cpp @@ -0,0 +1,106 @@ +#include "vdm/util/event_bus.hpp" + +#include +#include +#include +#include + +#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([&](const Progress &p) { + ++progress_hits; + VT_CHECK_EQ(p.task, 7); + }); + bus.subscribe([&](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 order; + bus.subscribe([&](const Progress &) { order.push_back(1); }); + bus.subscribe([&](const Progress &) { order.push_back(2); }); + bus.subscribe([&](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([&](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([&](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([&](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 total{0}; + bus.subscribe([&](const Progress &p) { total += p.downloaded; }); + + constexpr int kThreads = 8; + constexpr int kPerThread = 2000; + std::vector 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(kThreads) * kPerThread); +} diff --git a/core/tests/util/log_test.cpp b/core/tests/util/log_test.cpp new file mode 100644 index 0000000..c62e4e9 --- /dev/null +++ b/core/tests/util/log_test.cpp @@ -0,0 +1,64 @@ +#include "vdm/util/log.hpp" + +#include +#include +#include + +#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>(); + vdm::set_log_sink(std::make_shared([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(0); + vdm::set_log_sink(std::make_shared( + [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")); +} diff --git a/core/tests/util/result_test.cpp b/core/tests/util/result_test.cpp new file mode 100644 index 0000000..da6ce1c --- /dev/null +++ b/core/tests/util/result_test.cpp @@ -0,0 +1,108 @@ +#include "vdm/util/result.hpp" + +#include + +#include "vtest.hpp" + +using vdm::Err; +using vdm::Error; +using vdm::ErrorInfo; +using vdm::Result; + +VT_TEST(result_holds_value) { + Result r = 42; + VT_REQUIRE(r.has_value()); + VT_CHECK(static_cast(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 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 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 r = vdm::ok(); + VT_CHECK(r.has_value()); + VT_CHECK_EQ(r.code(), Error::ok); +} + +VT_TEST(result_void_error) { + Result 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 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 r = Err{Error::connection_reset, "peer RST"}; + int calls = 0; + auto next = r.and_then([&](int v) -> std::expected { + ++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 parse_positive(int n) { + if (n < 0) + return Err{Error::internal, "negative"}; + return n; +} + +Result 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)); +} diff --git a/core/tests/util/thread_pool_test.cpp b/core/tests/util/thread_pool_test.cpp new file mode 100644 index 0000000..b5ab28e --- /dev/null +++ b/core/tests/util/thread_pool_test.cpp @@ -0,0 +1,70 @@ +#include "vdm/util/thread_pool.hpp" + +#include +#include +#include +#include +#include +#include + +#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 sum{0}; + std::vector> 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 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 ran{false}; + auto f = pool.submit([&ran] { ran = true; }); + f.get(); + VT_CHECK(ran.load()); +}