merge: libveloxproto target and stage 4 io layer

core/CMakeLists.txt

@@ -1,8 +1,8 @@
-# libveloxcore — the download engine. Lane CORE.
-#
-# No JSON, no SQL, no Qt, no RPC in this tree (CLAUDE.md §3, AGENT-CORE brief).
-# This file is self-contained; it is wired into the build by PKG uncommenting
-# `add_subdirectory(core)` in the root CMakeLists.txt (see core/docs/pkg-requests-m1.md).
+# core/ produces TWO targets (ADR 0009):
+#   veloxcore  — the download engine. No JSON, no SQL, no Qt, no RPC. Ever (CLAUDE.md §3).
+#   veloxproto — the generated wire types, which ARE JSON. NOT linked by veloxcore.
+# The `no JSON in core/` rule constrains core/src/ and core/include/; core/generated/ is
+# the sanctioned exception. Wired in by PKG via add_subdirectory(core) in the root file.
 
 find_package(Threads REQUIRED)
 find_package(CURL 8.0 REQUIRED)
@@ -17,6 +17,8 @@ add_library(veloxcore STATIC
     src/net/content_disposition.cpp
     src/net/url.cpp
     src/net/probe.cpp
+    src/io/sparse_file.cpp
+    src/io/write_buffer.cpp
 )
 add_library(velox::core ALIAS veloxcore)
 
@@ -37,6 +39,32 @@ target_link_libraries(veloxcore PUBLIC Threads::Threads CURL::libcurl)
 
 # Later stages add: find_package(OpenSSL) for meta/ (streaming SHA-256 + resume CRC).
 
+# --- libveloxproto — generated wire code (ADR 0009) --------------------------------------
+# Its own target so libveloxcore stays JSON-free. Consumed by veloxd, the CLI, the GUI and
+# the conformance runner. The root CMakeLists only find_package(nlohmann_json)'s when
+# daemon/ has landed, so find it here too — this must build even if core is the only lane.
+if(NOT TARGET nlohmann_json::nlohmann_json)
+    find_package(nlohmann_json 3.11 REQUIRED)
+endif()
+
+add_library(veloxproto STATIC generated/velox_proto.cpp)
+add_library(velox::proto ALIAS veloxproto)
+
+target_include_directories(veloxproto PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/generated)
+target_compile_features(veloxproto PUBLIC cxx_std_23)
+target_link_libraries(veloxproto PUBLIC nlohmann_json::nlohmann_json)
+
+# Generated code is committed and never hand-edited (CLAUDE.md §2); do not fail the build
+# on a codegen quirk that trips -Werror. Warnings stay on for visibility.
+target_compile_options(veloxproto PRIVATE -Wall -Wextra -Wno-error)
+
+# A build-time tripwire for the split ADR 0009 exists to protect: veloxcore must never end
+# up linking veloxproto.
+get_target_property(_core_links veloxcore LINK_LIBRARIES)
+if(_core_links AND "veloxproto" IN_LIST _core_links)
+    message(FATAL_ERROR "veloxcore links veloxproto — ADR 0009 violation (engine sees JSON).")
+endif()
+
 if(VELOX_BUILD_TESTS)
     add_subdirectory(tests)
 endif()

core/docs/proto-requests-conformance-cmake.md

@@ -0,0 +1,69 @@
+# CORE → PROTO — `tests/conformance/cpp/CMakeLists.txt` should link `veloxproto` now
+
+Status: **open**. Small, mechanical. Filed rather than fixed because `tests/conformance/`
+is PROTO's lane.
+
+## What's stale
+
+`tests/conformance/cpp/CMakeLists.txt` says in its header comment:
+
+> Links libveloxproto (the generated protocol code in core/generated/), not libveloxcore
+
+…but it actually **compiles `core/generated/velox_proto.cpp` straight into the
+executable** and finds `nlohmann_json` itself:
+
+```cmake
+add_executable(velox_conformance_cpp
+    conformance_main.cpp
+    ${CMAKE_SOURCE_DIR}/core/generated/velox_proto.cpp)
+
+target_include_directories(velox_conformance_cpp PRIVATE
+    ${CMAKE_SOURCE_DIR}/core/generated
+    ${CMAKE_CURRENT_SOURCE_DIR})
+target_link_libraries(velox_conformance_cpp PRIVATE nlohmann_json::nlohmann_json)
+```
+
+That was the only option while ADR 0009's `libveloxproto` target didn't exist. **It exists
+now** — `core/CMakeLists.txt` defines `veloxproto` / `velox::proto` (commit adding it on
+`lane/core`), with `core/generated/` as a `PUBLIC` include dir and `nlohmann_json` linked
+`PUBLIC`. The comment and the code now agree only if the runner links the target.
+
+## Requested change
+
+```cmake
+if(TARGET velox::proto)
+    add_executable(velox_conformance_cpp conformance_main.cpp)
+    target_link_libraries(velox_conformance_cpp PRIVATE velox::proto)
+else()
+    # Standalone configure of tests/conformance/ (no core/ in the tree): fall back to
+    # compiling the generated source directly, as today.
+    if(NOT TARGET nlohmann_json::nlohmann_json)
+        find_package(nlohmann_json 3.11 REQUIRED)
+    endif()
+    add_executable(velox_conformance_cpp
+        conformance_main.cpp
+        ${CMAKE_SOURCE_DIR}/core/generated/velox_proto.cpp)
+    target_include_directories(velox_conformance_cpp PRIVATE
+        ${CMAKE_SOURCE_DIR}/core/generated)
+    target_link_libraries(velox_conformance_cpp PRIVATE nlohmann_json::nlohmann_json)
+endif()
+
+target_include_directories(velox_conformance_cpp PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+target_compile_features(velox_conformance_cpp PRIVATE cxx_std_23)
+add_test(NAME conformance_cpp COMMAND velox_conformance_cpp ${CMAKE_SOURCE_DIR})
+set_tests_properties(conformance_cpp PROPERTIES LABELS "conformance")
+```
+
+The `if(TARGET ...)` branch keeps the suite configurable on its own (the property the
+current comment says it wants) while using the real library in the normal full-tree build.
+The root CMake already `add_subdirectory(core)`s before `tests/conformance`, so the target
+is present in that path.
+
+## Why it matters beyond tidiness
+
+GUI is blocked on `libveloxproto` being a real link target (it can't `add_subdirectory` a
+sibling lane's `core/generated/` and re-guess the nlohmann find). Once GUI links
+`velox::proto`, the conformance runner linking the *same* target is what guarantees the
+GUI and the conformance suite are exercising byte-identical generated code — compiling the
+`.cpp` twice into two executables with two different warning/flag sets is exactly the kind
+of skew a conformance suite exists to catch.

core/include/vdm/io/sparse_file.hpp

@@ -0,0 +1,80 @@
+// vdm/io/sparse_file.hpp — the one output file, written at absolute offsets.
+//
+// docs/04 §4: one file, opened once, O_WRONLY; posix_fallocate the full size up front;
+// each segment pwrite()s at its own offset so there is no reassembly pass; fadvise
+// DONTNEED on written ranges; fdatasync on a timer, never per write.
+//
+// This header compiles standalone.
+
+#ifndef VDM_IO_SPARSE_FILE_HPP
+#define VDM_IO_SPARSE_FILE_HPP
+
+#include <cstdint>
+#include <string>
+#include <string_view>
+
+#include "vdm/util/bytes.hpp"
+#include "vdm/util/result.hpp"
+
+namespace vdm::io {
+
+class SparseFile {
+  public:
+    struct OpenOptions {
+        std::uint64_t total_size = 0;    // full final size; 0 = unknown (chunked transfer)
+        bool preallocate = true;         // posix_fallocate; falls back to ftruncate
+        bool truncate_existing = false;  // start fresh (true) vs. resume into an existing
+                                         // part file (false)
+    };
+
+    SparseFile() = default;
+    ~SparseFile();
+
+    SparseFile(SparseFile &&) noexcept;
+    SparseFile &operator=(SparseFile &&) noexcept;
+    SparseFile(const SparseFile &) = delete;
+    SparseFile &operator=(const SparseFile &) = delete;
+
+    // Open `path` for writing at absolute offsets, creating it if needed. With
+    // `preallocate` and a known `total_size`, posix_fallocate the whole file (contiguous
+    // extents, no ENOSPC surprise at 99%). EOPNOTSUPP/ENOSYS (tmpfs, some network FS)
+    // falls back to ftruncate — sparse, no extent reservation — and is reported by
+    // preallocated().
+    [[nodiscard]] Result<void> open(std::string_view path, const OpenOptions &opts);
+    [[nodiscard]] Result<void> open(std::string_view path);  // default OpenOptions
+
+    [[nodiscard]] bool is_open() const noexcept { return fd_ >= 0; }
+    [[nodiscard]] bool preallocated() const noexcept { return preallocated_; }
+    [[nodiscard]] std::string_view path() const noexcept { return path_; }
+
+    // pwrite the whole span at `offset`, looping over short writes and retrying EINTR.
+    // Safe to call concurrently with other write_at()/sync() on the same object as long
+    // as the byte ranges do not overlap — POSIX guarantees each pwrite is atomic for a
+    // regular file, so no lock is taken on the hot path.
+    [[nodiscard]] Result<void> write_at(std::uint64_t offset, ConstByteSpan data);
+
+    // fdatasync. Call on a timer (default 5 s) and on pause — never per write (docs/04 §4).
+    [[nodiscard]] Result<void> sync();
+
+    // POSIX_FADV_DONTNEED on [offset, offset+len): drop already-written pages from the
+    // page cache so a 40 GB ISO does not evict the user's working set. Most effective
+    // after sync(). Best-effort — failures are ignored.
+    void advise_dontneed(std::uint64_t offset, std::uint64_t len) noexcept;
+
+    // ftruncate to `size`: give a chunked download its real size once known, or trim a
+    // preallocated tail that a steal/mirror never filled.
+    [[nodiscard]] Result<void> resize(std::uint64_t size);
+
+    [[nodiscard]] Result<void> close();
+
+  private:
+    void reset() noexcept;
+
+    int fd_ = -1;
+    bool preallocated_ = false;
+    std::string path_;
+};
+
+}  // namespace vdm::io
+
+#endif  // VDM_IO_SPARSE_FILE_HPP

core/include/vdm/io/write_buffer.hpp

@@ -0,0 +1,69 @@
+// vdm/io/write_buffer.hpp — per-segment accumulate-and-flush buffer.
+//
+// docs/04 §4: curl's write callback appends; the buffer is flushed with a single pwrite
+// when full or when the segment ends. §8: NO ALLOCATION in the write-callback hot path —
+// the buffer is preallocated at construction and append() only memcpys.
+//
+// Single-threaded. The owning download_task (stage 8) adds the disk-writer-thread handoff
+// (double buffering) on top; this primitive takes no lock.
+//
+// This header compiles standalone.
+
+#ifndef VDM_IO_WRITE_BUFFER_HPP
+#define VDM_IO_WRITE_BUFFER_HPP
+
+#include <cstddef>
+#include <cstdint>
+#include <functional>
+#include <vector>
+
+#include "vdm/util/bytes.hpp"
+#include "vdm/util/result.hpp"
+
+namespace vdm::io {
+
+class WriteBuffer {
+  public:
+    // Writes `span` durably at absolute file offset `offset`. Must not allocate
+    // (SparseFile::write_at doesn't). Return an error to abort the segment; WriteBuffer
+    // propagates it and keeps the un-flushed bytes so the caller can decide.
+    using FlushFn = std::function<Result<void>(std::uint64_t offset, ConstByteSpan span)>;
+
+    // `capacity` is the effective per-segment buffer_bytes (already clamped by the
+    // segmenter against max_total_buffer_bytes). Must be > 0.
+    WriteBuffer(std::uint64_t start_offset, std::size_t capacity, FlushFn flush);
+
+    WriteBuffer(WriteBuffer &&) noexcept = default;
+    WriteBuffer &operator=(WriteBuffer &&) noexcept = default;
+    WriteBuffer(const WriteBuffer &) = delete;
+    WriteBuffer &operator=(const WriteBuffer &) = delete;
+
+    // Append body bytes. Flushes automatically each time the buffer fills. A chunk at
+    // least `capacity` bytes long, arriving when the buffer is empty, is written straight
+    // through (one extra flush call, still no allocation and no memcpy).
+    [[nodiscard]] Result<void> append(ConstByteSpan span);
+
+    // Flush whatever is buffered right now. Call at segment end and on pause. A no-op
+    // (success) when nothing is pending.
+    [[nodiscard]] Result<void> flush();
+
+    [[nodiscard]] std::size_t capacity() const noexcept { return buf_.size(); }
+    [[nodiscard]] std::size_t pending() const noexcept { return len_; }
+    // File offset the next flush will write at (start + all bytes already flushed).
+    [[nodiscard]] std::uint64_t next_offset() const noexcept { return base_; }
+    // Total bytes handed to append() over this buffer's life.
+    [[nodiscard]] std::uint64_t total_appended() const noexcept { return appended_; }
+
+  private:
+    Result<void> flush_pending();
+
+    std::vector<std::byte> buf_;
+    std::size_t len_ = 0;     // bytes currently in buf_
+    std::uint64_t base_ = 0;  // file offset of buf_[0]
+    std::uint64_t appended_ = 0;
+    FlushFn flush_;
+};
+
+}  // namespace vdm::io
+
+#endif  // VDM_IO_WRITE_BUFFER_HPP

core/src/io/sparse_file.cpp

@@ -0,0 +1,181 @@
+// vdm/io/sparse_file.cpp
+
+#include "vdm/io/sparse_file.hpp"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <cstring>
+#include <utility>
+
+namespace vdm::io {
+namespace {
+
+Error errno_to_error(int e) noexcept {
+    switch (e) {
+        case ENOSPC:
+        case EDQUOT:
+            return Error::disk_full;
+        case EACCES:
+        case EPERM:
+        case EROFS:
+            return Error::permission_denied;
+        case ENOENT:
+        case ENOTDIR:
+        case EISDIR:
+        case ENAMETOOLONG:
+        case ELOOP:
+            return Error::path_rejected;
+        default:
+            return Error::io_error;
+    }
+}
+
+ErrorInfo sys_error(std::string_view what, int e) {
+    return ErrorInfo(errno_to_error(e), std::string(what) + ": " + std::strerror(e));
+}
+
+}  // namespace
+
+SparseFile::~SparseFile() {
+    if (fd_ >= 0)
+        ::close(fd_);
+}
+
+SparseFile::SparseFile(SparseFile &&o) noexcept
+    : fd_(std::exchange(o.fd_, -1)),
+      preallocated_(std::exchange(o.preallocated_, false)),
+      path_(std::move(o.path_)) {}
+
+SparseFile &SparseFile::operator=(SparseFile &&o) noexcept {
+    if (this != &o) {
+        if (fd_ >= 0)
+            ::close(fd_);
+        fd_ = std::exchange(o.fd_, -1);
+        preallocated_ = std::exchange(o.preallocated_, false);
+        path_ = std::move(o.path_);
+    }
+    return *this;
+}
+
+void SparseFile::reset() noexcept {
+    fd_ = -1;
+    preallocated_ = false;
+    path_.clear();
+}
+
+Result<void> SparseFile::open(std::string_view path) {
+    return open(path, OpenOptions{});
+}
+
+Result<void> SparseFile::open(std::string_view path, const OpenOptions &opts) {
+    if (fd_ >= 0)
+        return ErrorInfo(Error::internal, "SparseFile already open");
+
+    std::string p(path);
+    int flags = O_WRONLY | O_CREAT | O_CLOEXEC;
+    if (opts.truncate_existing)
+        flags |= O_TRUNC;
+
+    int fd = ::open(p.c_str(), flags, 0644);
+    if (fd < 0)
+        return sys_error("open " + p, errno);
+
+    bool prealloc = false;
+    if (opts.total_size > 0) {
+        if (opts.preallocate) {
+            // posix_fallocate returns the error number directly and does not set errno.
+            int rc = ::posix_fallocate(fd, 0, static_cast<off_t>(opts.total_size));
+            if (rc == 0) {
+                prealloc = true;
+            } else if (rc == EOPNOTSUPP || rc == ENOSYS || rc == EINVAL) {
+                if (::ftruncate(fd, static_cast<off_t>(opts.total_size)) != 0) {
+                    int e = errno;
+                    ::close(fd);
+                    return sys_error("ftruncate " + p, e);
+                }
+            } else {
+                ::close(fd);
+                return sys_error("posix_fallocate " + p, rc);
+            }
+        } else if (!opts.truncate_existing) {
+            // Resuming: make sure the file is at least total_size so pwrite offsets land.
+            if (::ftruncate(fd, static_cast<off_t>(opts.total_size)) != 0) {
+                int e = errno;
+                ::close(fd);
+                return sys_error("ftruncate " + p, e);
+            }
+        }
+    }
+
+    fd_ = fd;
+    preallocated_ = prealloc;
+    path_ = std::move(p);
+    return ok();
+}
+
+Result<void> SparseFile::write_at(std::uint64_t offset, ConstByteSpan data) {
+    if (fd_ < 0)
+        return ErrorInfo(Error::internal, "write_at on a closed SparseFile");
+
+    const std::byte *p = data.data();
+    std::size_t remaining = data.size();
+    off_t pos = static_cast<off_t>(offset);
+
+    while (remaining > 0) {
+        ssize_t n = ::pwrite(fd_, p, remaining, pos);
+        if (n < 0) {
+            if (errno == EINTR)
+                continue;
+            return sys_error("pwrite", errno);
+        }
+        if (n == 0)
+            return ErrorInfo(Error::io_error, "pwrite returned 0");
+        p += n;
+        pos += n;
+        remaining -= static_cast<std::size_t>(n);
+    }
+    return ok();
+}
+
+Result<void> SparseFile::sync() {
+    if (fd_ < 0)
+        return ErrorInfo(Error::internal, "sync on a closed SparseFile");
+    while (::fdatasync(fd_) != 0) {
+        if (errno == EINTR)
+            continue;
+        return sys_error("fdatasync", errno);
+    }
+    return ok();
+}
+
+void SparseFile::advise_dontneed(std::uint64_t offset, std::uint64_t len) noexcept {
+    if (fd_ < 0 || len == 0)
+        return;
+    ::posix_fadvise(fd_, static_cast<off_t>(offset), static_cast<off_t>(len), POSIX_FADV_DONTNEED);
+}
+
+Result<void> SparseFile::resize(std::uint64_t size) {
+    if (fd_ < 0)
+        return ErrorInfo(Error::internal, "resize on a closed SparseFile");
+    while (::ftruncate(fd_, static_cast<off_t>(size)) != 0) {
+        if (errno == EINTR)
+            continue;
+        return sys_error("ftruncate", errno);
+    }
+    return ok();
+}
+
+Result<void> SparseFile::close() {
+    if (fd_ < 0)
+        return ok();
+    int fd = std::exchange(fd_, -1);
+    int rc = ::close(fd);
+    reset();
+    if (rc != 0)
+        return sys_error("close", errno);
+    return ok();
+}
+
+}  // namespace vdm::io

core/src/io/write_buffer.cpp

@@ -0,0 +1,57 @@
+// vdm/io/write_buffer.cpp
+
+#include "vdm/io/write_buffer.hpp"
+
+#include <algorithm>
+#include <cassert>
+#include <cstring>
+#include <utility>
+
+namespace vdm::io {
+
+WriteBuffer::WriteBuffer(std::uint64_t start_offset, std::size_t capacity, FlushFn flush)
+    : buf_(capacity), base_(start_offset), flush_(std::move(flush)) {
+    assert(capacity > 0 && "WriteBuffer capacity must be > 0");
+}
+
+Result<void> WriteBuffer::flush_pending() {
+    if (len_ == 0)
+        return ok();
+    VDM_TRY(flush_(base_, ConstByteSpan(buf_.data(), len_)));
+    base_ += len_;
+    len_ = 0;
+    return ok();
+}
+
+Result<void> WriteBuffer::append(ConstByteSpan span) {
+    // On an error return, next_offset() still reflects exactly what is durable; buffered
+    // (non-durable) bytes and any unconsumed tail of `span` are the caller's to abandon —
+    // the segment aborts and resumes/restarts from next_offset().
+    while (!span.empty()) {
+        // Buffer empty and the incoming chunk fills at least a whole buffer: write it
+        // straight through — no memcpy, no allocation, just an extra flush call.
+        if (len_ == 0 && span.size() >= buf_.size()) {
+            VDM_TRY(flush_(base_, span));
+            base_ += span.size();
+            appended_ += span.size();
+            return ok();
+        }
+
+        const std::size_t room = buf_.size() - len_;
+        const std::size_t n = std::min(span.size(), room);
+        std::memcpy(buf_.data() + len_, span.data(), n);
+        len_ += n;
+        appended_ += n;
+        span = span.subspan(n);
+
+        if (len_ == buf_.size())
+            VDM_TRY(flush_pending());
+    }
+    return ok();
+}
+
+Result<void> WriteBuffer::flush() {
+    return flush_pending();
+}
+
+}  // namespace vdm::io

core/tests/CMakeLists.txt

@@ -25,6 +25,8 @@ vdm_add_test(veloxcore_log_test       util/log_test.cpp)
 # in the tree yet (lanes merge independently).
 vdm_add_test(veloxcore_content_disposition_test net/content_disposition_test.cpp)
 vdm_add_test(veloxcore_url_test net/url_test.cpp)
+vdm_add_test(veloxcore_sparse_file_test io/sparse_file_test.cpp)
+vdm_add_test(veloxcore_write_buffer_test io/write_buffer_test.cpp)
 
 set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py)
 foreach(net_it http_client probe)

core/tests/io/sparse_file_test.cpp

@@ -0,0 +1,181 @@
+#include "vdm/io/sparse_file.hpp"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <array>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "vtest.hpp"
+
+using vdm::Error;
+using vdm::io::SparseFile;
+
+namespace {
+
+struct TempPath {
+    std::string path;
+    TempPath() {
+        const char *dir = std::getenv("TMPDIR");
+        path = (dir ? dir : "/tmp");
+        path += "/vdm_sparse_test_XXXXXX";
+        int fd = ::mkstemp(path.data());
+        if (fd >= 0) {
+            ::close(fd);
+            ::unlink(path.c_str());  // we only wanted a unique name
+        }
+    }
+    ~TempPath() { ::unlink(path.c_str()); }
+};
+
+std::vector<std::byte> read_all(const std::string &path) {
+    int fd = ::open(path.c_str(), O_RDONLY);
+    if (fd < 0)
+        return {};
+    std::vector<std::byte> out;
+    std::array<std::byte, 4096> buf{};
+    for (;;) {
+        ssize_t n = ::read(fd, buf.data(), buf.size());
+        if (n <= 0)
+            break;
+        out.insert(out.end(), buf.begin(), buf.begin() + n);
+    }
+    ::close(fd);
+    return out;
+}
+
+std::uint64_t file_size(const std::string &path) {
+    int fd = ::open(path.c_str(), O_RDONLY);
+    if (fd < 0)
+        return 0;
+    off_t end = ::lseek(fd, 0, SEEK_END);
+    ::close(fd);
+    return end < 0 ? 0 : static_cast<std::uint64_t>(end);
+}
+
+vdm::ConstByteSpan bytes(const char *s) {
+    return {reinterpret_cast<const std::byte *>(s), std::strlen(s)};
+}
+
+}  // namespace
+
+VT_TEST(sparse_open_preallocates_full_size) {
+    TempPath tp;
+    SparseFile f;
+    auto r = f.open(tp.path, {.total_size = 1 << 20});
+    VT_REQUIRE(r.has_value());
+    VT_CHECK(f.is_open());
+    VT_CHECK_EQ(file_size(tp.path), 1u << 20);
+    // /tmp is usually a real fs; if it's tmpfs, preallocated() is false and that's fine.
+    VT_CHECK(f.close().has_value());
+}
+
+VT_TEST(sparse_write_at_absolute_offsets) {
+    TempPath tp;
+    SparseFile f;
+    VT_REQUIRE(f.open(tp.path, {.total_size = 64}).has_value());
+
+    VT_CHECK(f.write_at(10, bytes("hello")).has_value());
+    VT_CHECK(f.write_at(40, bytes("world")).has_value());
+    VT_CHECK(f.sync().has_value());
+
+    auto data = read_all(tp.path);
+    VT_REQUIRE(data.size() == 64);
+    VT_CHECK_EQ(std::memcmp(data.data() + 10, "hello", 5), 0);
+    VT_CHECK_EQ(std::memcmp(data.data() + 40, "world", 5), 0);
+    f.close().value();
+}
+
+VT_TEST(sparse_write_past_end_grows_file) {
+    TempPath tp;
+    SparseFile f;
+    VT_REQUIRE(f.open(tp.path, {.total_size = 16}).has_value());
+    VT_CHECK(f.write_at(1000, bytes("tail")).has_value());
+    VT_CHECK_EQ(file_size(tp.path), 1004u);
+    f.close().value();
+}
+
+VT_TEST(sparse_resize_trims) {
+    TempPath tp;
+    SparseFile f;
+    VT_REQUIRE(f.open(tp.path, {.total_size = 4096}).has_value());
+    VT_CHECK(f.resize(100).has_value());
+    VT_CHECK_EQ(file_size(tp.path), 100u);
+    f.close().value();
+}
+
+VT_TEST(sparse_open_bad_path_is_path_rejected) {
+    SparseFile f;
+    auto r = f.open("/vdm_no_such_dir_xyz/file.part", {.total_size = 10});
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::path_rejected);
+    VT_CHECK(!f.is_open());
+}
+
+VT_TEST(sparse_ops_on_closed_file_error) {
+    SparseFile f;
+    VT_CHECK_EQ(f.write_at(0, bytes("x")).error().code, Error::internal);
+    VT_CHECK_EQ(f.sync().error().code, Error::internal);
+    VT_CHECK(f.close().has_value());  // close on a closed file is ok
+}
+
+VT_TEST(sparse_advise_dontneed_is_safe) {
+    TempPath tp;
+    SparseFile f;
+    VT_REQUIRE(f.open(tp.path, {.total_size = 8192}).has_value());
+    VT_CHECK(f.write_at(0, bytes("data")).has_value());
+    VT_CHECK(f.sync().has_value());
+    f.advise_dontneed(0, 4096);  // must not crash / must be a no-op-safe call
+    f.advise_dontneed(0, 0);
+    f.close().value();
+}
+
+VT_TEST(sparse_move_transfers_fd) {
+    TempPath tp;
+    SparseFile a;
+    VT_REQUIRE(a.open(tp.path, {.total_size = 32}).has_value());
+    SparseFile b = std::move(a);
+    VT_CHECK(!a.is_open());
+    VT_CHECK(b.is_open());
+    VT_CHECK(b.write_at(0, bytes("moved")).has_value());
+    b.close().value();
+}
+
+VT_TEST(sparse_concurrent_nonoverlapping_writes) {
+    TempPath tp;
+    SparseFile f;
+    constexpr int kSegs = 8;
+    constexpr std::size_t kSeg = 64 * 1024;
+    VT_REQUIRE(f.open(tp.path, {.total_size = kSegs * kSeg}).has_value());
+
+    std::vector<std::jthread> ts;
+    for (int s = 0; s < kSegs; ++s) {
+        ts.emplace_back([&, s] {
+            std::vector<std::byte> chunk(kSeg, static_cast<std::byte>('A' + s));
+            for (std::size_t off = 0; off < kSeg; off += 4096) {
+                auto r = f.write_at(static_cast<std::uint64_t>(s) * kSeg + off,
+                                    vdm::ConstByteSpan(chunk.data() + off, 4096));
+                if (!r.has_value())
+                    VT_FAIL("concurrent write_at failed");
+            }
+        });
+    }
+    ts.clear();  // join
+    VT_CHECK(f.sync().has_value());
+
+    auto data = read_all(tp.path);
+    VT_REQUIRE(data.size() == kSegs * kSeg);
+    for (int s = 0; s < kSegs; ++s) {
+        bool ok = true;
+        for (std::size_t i = 0; i < kSeg; ++i)
+            if (data[s * kSeg + i] != static_cast<std::byte>('A' + s))
+                ok = false;
+        VT_CHECK(ok);
+    }
+    f.close().value();
+}

core/tests/io/write_buffer_test.cpp

@@ -0,0 +1,202 @@
+#include "vdm/io/write_buffer.hpp"
+
+#include <atomic>
+#include <cstdlib>
+#include <cstring>
+#include <new>
+#include <string>
+#include <vector>
+
+#include "vtest.hpp"
+
+using vdm::ConstByteSpan;
+using vdm::Error;
+using vdm::Result;
+using vdm::io::WriteBuffer;
+
+// --- global allocation counter, for the "no alloc in append()" test -----------------
+namespace {
+std::atomic<long> g_alloc_calls{0};
+std::atomic<bool> g_count_allocs{false};
+}  // namespace
+
+void *operator new(std::size_t n) {
+    if (g_count_allocs.load(std::memory_order_relaxed))
+        g_alloc_calls.fetch_add(1, std::memory_order_relaxed);
+    void *p = std::malloc(n ? n : 1);
+    if (!p)
+        throw std::bad_alloc();
+    return p;
+}
+void operator delete(void *p) noexcept {
+    std::free(p);
+}
+void operator delete(void *p, std::size_t) noexcept {
+    std::free(p);
+}
+void *operator new[](std::size_t n) {
+    return ::operator new(n);
+}
+void operator delete[](void *p) noexcept {
+    std::free(p);
+}
+void operator delete[](void *p, std::size_t) noexcept {
+    std::free(p);
+}
+
+namespace {
+
+// A flush sink that records (offset, bytes) and never allocates after construction.
+struct Sink {
+    std::vector<std::byte> data;      // pre-reserved
+    std::vector<std::uint64_t> offs;  // pre-reserved
+    std::vector<std::size_t> lens;
+    bool fail_next = false;
+
+    WriteBuffer::FlushFn fn() {
+        return [this](std::uint64_t off, ConstByteSpan s) -> Result<void> {
+            if (fail_next) {
+                fail_next = false;
+                return vdm::Err{Error::io_error, "sink forced failure"};
+            }
+            offs.push_back(off);
+            lens.push_back(s.size());
+            data.insert(data.end(), s.begin(), s.end());
+            return vdm::ok();
+        };
+    }
+};
+
+ConstByteSpan sv(const char *s) {
+    return {reinterpret_cast<const std::byte *>(s), std::strlen(s)};
+}
+
+}  // namespace
+
+VT_TEST(wb_accumulates_then_flushes_on_fill) {
+    Sink sink;
+    sink.data.reserve(1 << 16);
+    sink.offs.reserve(64);
+    sink.lens.reserve(64);
+
+    WriteBuffer wb(0, 8, sink.fn());
+    VT_CHECK(wb.append(sv("abc")).has_value());  // 3 buffered
+    VT_CHECK_EQ(wb.pending(), 3u);
+    VT_CHECK(sink.offs.empty());  // no flush yet
+
+    VT_CHECK(wb.append(sv("defgh")).has_value());  // fills to 8 -> flush
+    VT_REQUIRE(sink.offs.size() == 1);
+    VT_CHECK_EQ(sink.offs[0], 0u);
+    VT_CHECK_EQ(sink.lens[0], 8u);
+    VT_CHECK_EQ(wb.pending(), 0u);
+    VT_CHECK_EQ(wb.next_offset(), 8u);
+
+    VT_CHECK(wb.append(sv("ij")).has_value());
+    VT_CHECK(wb.flush().has_value());  // explicit tail flush
+    VT_REQUIRE(sink.offs.size() == 2);
+    VT_CHECK_EQ(sink.offs[1], 8u);
+    VT_CHECK_EQ(sink.lens[1], 2u);
+
+    VT_CHECK_EQ(std::string(reinterpret_cast<const char *>(sink.data.data()), sink.data.size()),
+                std::string("abcdefghij"));
+    VT_CHECK_EQ(wb.total_appended(), 10u);
+}
+
+VT_TEST(wb_flush_is_noop_when_empty) {
+    Sink sink;
+    sink.offs.reserve(4);
+    WriteBuffer wb(100, 16, sink.fn());
+    VT_CHECK(wb.flush().has_value());
+    VT_CHECK(sink.offs.empty());
+}
+
+VT_TEST(wb_oversized_chunk_writes_through) {
+    Sink sink;
+    sink.data.reserve(1 << 16);
+    sink.offs.reserve(16);
+    sink.lens.reserve(16);
+
+    WriteBuffer wb(0, 8, sink.fn());
+    VT_CHECK(wb.append(sv("ab")).has_value());  // 2 buffered
+    // 20 bytes arriving: buffer isn't empty, so first 6 top it off + flush(8), then the
+    // remaining 14 (>= capacity, buffer now empty) write straight through.
+    std::string big(20, 'x');
+    VT_CHECK(wb.append(sv(big.c_str())).has_value());
+    VT_CHECK(wb.flush().has_value());
+
+    // reconstruct
+    std::string got(reinterpret_cast<const char *>(sink.data.data()), sink.data.size());
+    VT_CHECK_EQ(got, std::string("ab") + big);
+    VT_CHECK_EQ(wb.total_appended(), 22u);
+    // one full-buffer flush + one passthrough; order preserved
+    VT_CHECK(sink.offs.size() >= 2);
+    VT_CHECK_EQ(sink.offs.front(), 0u);
+}
+
+VT_TEST(wb_exact_capacity_chunk_from_empty_writes_through) {
+    Sink sink;
+    sink.data.reserve(64);
+    sink.offs.reserve(4);
+    sink.lens.reserve(4);
+    WriteBuffer wb(0, 4, sink.fn());
+    VT_CHECK(wb.append(sv("wxyz")).has_value());  // == capacity, empty -> passthrough
+    VT_REQUIRE(sink.offs.size() == 1);
+    VT_CHECK_EQ(sink.lens[0], 4u);
+    VT_CHECK_EQ(wb.pending(), 0u);
+}
+
+VT_TEST(wb_flush_error_propagates_without_advancing_durable_offset) {
+    Sink sink;
+    sink.data.reserve(64);
+    sink.offs.reserve(4);
+    sink.lens.reserve(4);
+    WriteBuffer wb(0, 8, sink.fn());
+
+    VT_CHECK(wb.append(sv("abc")).has_value());  // 3 buffered, nothing durable yet
+    VT_CHECK_EQ(wb.next_offset(), 0u);
+
+    sink.fail_next = true;
+    auto r = wb.flush();  // forced failure
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::io_error);
+    VT_CHECK_EQ(wb.next_offset(), 0u);  // durable offset did NOT move
+    VT_CHECK(sink.offs.empty());
+
+    // a retry flush succeeds and advances
+    VT_CHECK(wb.flush().has_value());
+    VT_CHECK_EQ(wb.next_offset(), 3u);
+    VT_REQUIRE(sink.lens.size() == 1);
+    VT_CHECK_EQ(sink.lens[0], 3u);
+}
+
+VT_TEST(wb_append_does_not_allocate) {
+    // flush sink that never allocates: just sum sizes.
+    std::atomic<std::uint64_t> total{0};
+    auto flush = [&total](std::uint64_t, ConstByteSpan s) -> Result<void> {
+        total.fetch_add(s.size());
+        return vdm::ok();
+    };
+    WriteBuffer wb(0, 4096, flush);
+
+    // Warm up (any first-call lazy init happens now, outside the measured window).
+    std::string warm(100, 'w');
+    (void)wb.append(sv(warm.c_str()));
+    (void)wb.flush();
+
+    std::vector<std::byte> chunk(512, std::byte{7});
+    g_alloc_calls.store(0);
+    g_count_allocs.store(true);
+    for (int i = 0; i < 5000; ++i) {
+        auto r = wb.append(ConstByteSpan(chunk.data(), 137 + (i % 200)));
+        if (!r.has_value()) {
+            g_count_allocs.store(false);
+            VT_FAIL("append failed");
+            return;
+        }
+    }
+    (void)wb.flush();
+    g_count_allocs.store(false);
+
+    VT_CHECK_EQ(g_alloc_calls.load(), 0L);
+    VT_CHECK(total.load() > 0);
+}

core/tests/support/vtest.hpp

@@ -129,11 +129,14 @@ inline int run_all() {
             ::vt::report(__FILE__, __LINE__, #COND, {}, /*fatal=*/true); \
     } while (0)
 
+// NOTE: operands are copied (auto, not auto&&). A test assertion must never outlive a
+// temporary the expression returned a reference into — the copy makes that safe. All
+// compared types here are cheap to copy.
 #define VT_CHECK_EQ(A, B)                                                  \
     do {                                                                   \
         ::vt::stats().checks++;                                            \
-        auto &&_a = (A);                                                   \
-        auto &&_b = (B);                                                   \
+        auto _a = (A);                                                     \
+        auto _b = (B);                                                     \
         if (!(_a == _b))                                                   \
             ::vt::report(__FILE__, __LINE__, #A " == " #B,                 \
                          ::vt::show(_a) + " vs " + ::vt::show(_b), false); \
@@ -142,8 +145,8 @@ inline int run_all() {
 #define VT_CHECK_NE(A, B)                                                  \
     do {                                                                   \
         ::vt::stats().checks++;                                            \
-        auto &&_a = (A);                                                   \
-        auto &&_b = (B);                                                   \
+        auto _a = (A);                                                     \
+        auto _b = (B);                                                     \
         if (!(_a != _b))                                                   \
             ::vt::report(__FILE__, __LINE__, #A " != " #B,                 \
                          ::vt::show(_a) + " vs " + ::vt::show(_b), false); \