ca-checkpoint.H

Go to the documentation of this file.

/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
 
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
#ifndef CA_CHECKPOINT_H
#define CA_CHECKPOINT_H
 
#include <array>
#include <atomic>
#include <cerrno>
#include <chrono>
#include <concepts>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <deque>
#include <filesystem>
#include <fstream>
#include <functional>
#include <iosfwd>
#include <mutex>
#include <optional>
#include <random>
#include <string>
#include <string_view>
#include <system_error>
#include <thread>
#include <type_traits>
#include <typeinfo>
#include <utility>
#include <vector>
 
#if defined(_WIN32)
#  include <io.h>
#  include <windows.h>
#else
#  include <fcntl.h>
#  include <sys/stat.h>
#  include <sys/types.h>
#  include <unistd.h>
#endif
 
#include <ah-errors.H>
 
#include <ca-checkpoint-compress.H>
#include <ca-rng.H>
#include <ca-traits.H>
#include <tpl_ca_concepts.H>
 
namespace Aleph {
namespace CA {
 
namespace ca_checkpoint_detail {
 
inline constexpr std::array<char, 8> magic_bytes
  = {{'A', 'L', 'E', 'P', 'H', 'C', 'A', '1'}};
 
inline constexpr std::uint32_t format_version = 2;
 
inline constexpr std::uint32_t format_version_min_read = 1;
 
inline constexpr std::streamoff header_bytes_v1
  = 8 + 4 + 8 + 4 + 4 + 4 * 8 + 8 + 1 + 7 + 8 + 8;  // 92
 
inline constexpr std::streamoff header_bytes_v2 = header_bytes_v1 + 24;  // 116
 
[[nodiscard]] inline constexpr std::uint64_t fnv1a_64(std::string_view s) noexcept
{
  std::uint64_t h = 0xcbf29ce484222325ull;
  for (const char c : s)
    {
      h ^= static_cast<std::uint64_t>(static_cast<unsigned char>(c));
      h *= 0x100000001b3ull;
    }
  return h;
}
 
template <typename T>
[[nodiscard]] inline std::uint64_t type_hash() noexcept
{
  const std::uint64_t name_hash = fnv1a_64(typeid(T).name());
  return splitmix64(name_hash ^ static_cast<std::uint64_t>(sizeof(T)));
}
 
// --- Little-endian byte serialisation --------------------------------
 
template <typename T>
inline void write_le(std::ostream &out, const T &value)
{
  static_assert(std::is_trivially_copyable_v<T>,
                "write_le requires a trivially copyable type");
  // We always emit the host's byte representation; for the integral
  // sizes we use it is identical to little-endian on every platform
  // Aleph-w supports.
  std::array<char, sizeof(T)> bytes{};
  std::memcpy(bytes.data(), &value, sizeof(T));
  out.write(bytes.data(), sizeof(T));
}
 
template <typename T>
[[nodiscard]] inline T read_le(std::istream &in)
{
  static_assert(std::is_trivially_copyable_v<T>,
                "read_le requires a trivially copyable type");
  std::array<char, sizeof(T)> bytes{};
  in.read(bytes.data(), sizeof(T));
  T value{};
  std::memcpy(&value, bytes.data(), sizeof(T));
  return value;
}
 
// --- Atomic-write primitives (Phase 17) ------------------------------
 
[[nodiscard]] inline std::filesystem::path
make_tmp_path(const std::filesystem::path &target)
{
  std::mt19937_64 rng(static_cast<std::uint64_t>(
    std::chrono::steady_clock::now().time_since_epoch().count())
                      ^ static_cast<std::uint64_t>(
                        reinterpret_cast<std::uintptr_t>(&target)));
  const std::uint64_t r = rng();
#if defined(_WIN32)
  const long pid = static_cast<long>(::GetCurrentProcessId());
#else
  const long pid = static_cast<long>(::getpid());
#endif
  std::string suffix = ".tmp." + std::to_string(pid) + "." + std::to_string(r);
  std::filesystem::path tmp = target;
  tmp += suffix;
  return tmp;
}
 
inline void sync_file(const std::filesystem::path &path)
{
#if defined(_WIN32)
  HANDLE h = ::CreateFileW(path.wstring().c_str(),
                           GENERIC_WRITE,
                           FILE_SHARE_READ,
                           nullptr,
                           OPEN_EXISTING,
                           FILE_ATTRIBUTE_NORMAL,
                           nullptr);
  ah_runtime_error_if(h == INVALID_HANDLE_VALUE)
    << "sync_file: CreateFileW failed for '" << path.string() << "'";
  const BOOL ok = ::FlushFileBuffers(h);
  const DWORD err = ok ? 0 : ::GetLastError();
  ::CloseHandle(h);
  ah_runtime_error_if(not ok)
    << "sync_file: FlushFileBuffers failed for '" << path.string()
    << "' (error " << err << ")";
#else
  const int fd = ::open(path.c_str(), O_WRONLY);
  ah_runtime_error_if(fd < 0)
    << "sync_file: open() failed for '" << path.string()
    << "' (errno=" << errno << ")";
  const int rc = ::fsync(fd);
  const int saved = errno;
  ::close(fd);
  ah_runtime_error_if(rc != 0)
    << "sync_file: fsync() failed for '" << path.string()
    << "' (errno=" << saved << ")";
#endif
}
 
inline void sync_directory(const std::filesystem::path &dir) noexcept
{
#if defined(_WIN32)
  (void) dir;
#else
  const int fd = ::open(dir.c_str(), O_RDONLY | O_DIRECTORY);
  if (fd < 0)
    return;
  (void) ::fsync(fd);
  ::close(fd);
#endif
}
 
class Tmp_File_Guard
{
  std::filesystem::path path_;
  bool committed_ = false;
 
public:
  explicit Tmp_File_Guard(std::filesystem::path p) noexcept
    : path_(std::move(p))
  {}
 
  Tmp_File_Guard(const Tmp_File_Guard &) = delete;
  Tmp_File_Guard &operator=(const Tmp_File_Guard &) = delete;
 
  ~Tmp_File_Guard() noexcept
  {
    if (not committed_)
      {
        std::error_code ec;
        std::filesystem::remove(path_, ec);
      }
  }
 
  void commit() noexcept { committed_ = true; }
 
  [[nodiscard]] const std::filesystem::path &path() const noexcept { return path_; }
};
 
}  // namespace ca_checkpoint_detail
 
// =====================================================================
// Public types
// =====================================================================
 
namespace ca_checkpoint_flags {
 
inline constexpr std::uint32_t compressed = 0x1;
 
inline constexpr std::uint32_t delta = 0x2;
 
}  // namespace ca_checkpoint_flags
 
struct Checkpoint_Header
{
  std::array<char, 8> magic{};
  std::uint32_t format_version = 0;
  std::uint64_t engine_type_hash = 0;
  std::uint32_t state_type_size = 0;
  std::uint32_t rank = 0;
  std::array<std::uint64_t, 4> extents{};
  std::uint64_t step_count = 0;
  std::uint8_t has_rng = 0;
  std::uint64_t master_seed = 0;
  std::uint64_t cell_count = 0;
 
  // --- v2-only trailer fields (default to 0 when reading v1) --------
 
  std::uint32_t flags = 0;
  std::uint32_t compression_level = 0;
  std::uint64_t payload_size = 0;
  std::uint64_t delta_base_step = 0;
};
 
struct Checkpoint_Options
{
  bool compress = false;
  int level = 6;
  bool sync_dir = true;
};
 
template <typename R>
concept Has_Master_Seed = requires(R &r, std::uint64_t s) {
  { r.master_seed() } -> std::convertible_to<std::uint64_t>;
  r.set_master_seed(s);
};
 
template <typename L>
concept Checkpointable_Lattice
  = LatticeLike<L> and requires(const L &l) {
                         { l.extents() } -> std::convertible_to<typename L::extents_type>;
                       };
 
template <typename E>
concept Checkpointable_Engine
  = Checkpointable_Lattice<typename E::lattice_type>
    and requires { typename E::rule_type; }
    and requires(E &e, std::size_t s) {
          { e.frame() } -> std::convertible_to<const typename E::lattice_type &>;
          { e.steps_run() } -> std::convertible_to<std::size_t>;
          { e.rule() } -> std::convertible_to<typename E::rule_type &>;
          { e.write_lattice() } -> std::convertible_to<typename E::lattice_type &>;
          e.set_step_count(s);
        };
 
struct Resume_Token
{
  std::filesystem::path source_path;
  Checkpoint_Header header{};
};
 
// =====================================================================
// inspect_checkpoint
// =====================================================================
 
[[nodiscard]] inline Checkpoint_Header
inspect_checkpoint(const std::filesystem::path &path)
{
  using namespace ca_checkpoint_detail;
  std::ifstream in(path, std::ios::binary);
  ah_runtime_error_if(not in)
    << "inspect_checkpoint: cannot open '" << path.string() << "'";
 
  Checkpoint_Header h;
  in.read(h.magic.data(), h.magic.size());
  ah_runtime_error_if(not in or h.magic != magic_bytes)
    << "inspect_checkpoint: bad magic in '" << path.string() << "'";
 
  h.format_version = read_le<std::uint32_t>(in);
  ah_runtime_error_if(h.format_version < format_version_min_read
                      or h.format_version > format_version)
    << "inspect_checkpoint: unsupported format version " << h.format_version
    << " (supported [" << format_version_min_read << ", " << format_version
    << "]) in '" << path.string() << "'";
 
  h.engine_type_hash = read_le<std::uint64_t>(in);
  h.state_type_size = read_le<std::uint32_t>(in);
  h.rank = read_le<std::uint32_t>(in);
  // save_checkpoint and load_checkpoint_into currently implement
  // rank-1, rank-2 and rank-3 serialisation branches; rank-4 is
  // intentionally rejected here so that a corrupted or future-version
  // header cannot bypass the save/load symmetry.
  ah_runtime_error_if(h.rank == 0 or h.rank > 3)
    << "inspect_checkpoint: rank " << h.rank << " out of range in '"
    << path.string() << "'";
  for (std::size_t d = 0; d < h.extents.size(); ++d)
    h.extents[d] = read_le<std::uint64_t>(in);
  h.step_count = read_le<std::uint64_t>(in);
  h.has_rng = read_le<std::uint8_t>(in);
  // 7 bytes of padding to keep the next field 8-aligned in the file.
  for (int i = 0; i < 7; ++i)
    (void) read_le<std::uint8_t>(in);
  h.master_seed = read_le<std::uint64_t>(in);
  h.cell_count = read_le<std::uint64_t>(in);
 
  if (h.format_version >= 2)
    {
      h.flags = read_le<std::uint32_t>(in);
      h.compression_level = read_le<std::uint32_t>(in);
      h.payload_size = read_le<std::uint64_t>(in);
      h.delta_base_step = read_le<std::uint64_t>(in);
    }
  else
    {
      h.flags = 0;
      h.compression_level = 0;
      h.payload_size = h.cell_count * h.state_type_size;
      h.delta_base_step = 0;
    }
 
  ah_runtime_error_if(not in)
    << "inspect_checkpoint: short read while parsing header of '"
    << path.string() << "'";
 
  // --- Hardening against hostile/corrupt headers (CWE-789, CWE-190) ---
  // Every field above is attacker-controlled. Validate internal
  // consistency here, once, so no downstream consumer can be driven into
  // an out-of-bounds or excessively large allocation:
  //   * cell_count must equal the product of the declared extents
  //     (this is an invariant of every file save_checkpoint writes), and
  //     that product must not overflow;
  //   * cell_count * state_type_size (the raw frame size) must not
  //     overflow size_t.
  std::uint64_t expected_cells = 1;
  for (std::uint32_t d = 0; d < h.rank; ++d)
    {
      const std::uint64_t e = h.extents[d];
      ah_runtime_error_if(e != 0 and expected_cells > UINT64_MAX / e)
        << "inspect_checkpoint: extent product overflows in '"
        << path.string() << "'";
      expected_cells *= e;
    }
  ah_runtime_error_if(h.cell_count != expected_cells)
    << "inspect_checkpoint: cell_count (" << h.cell_count
    << ") inconsistent with the product of extents (" << expected_cells
    << ") in '" << path.string() << "'";
  ah_runtime_error_if(h.state_type_size != 0
                      and h.cell_count > UINT64_MAX / h.state_type_size)
    << "inspect_checkpoint: cell_count * state_type_size overflows in '"
    << path.string() << "'";
 
  return h;
}
 
// =====================================================================
// Frame serialisation helpers (shared by sync / async / delta paths)
// =====================================================================
 
namespace ca_checkpoint_detail {
 
template <typename Engine>
[[nodiscard]] std::vector<std::uint8_t> snapshot_frame_bytes(const Engine &engine)
  requires Checkpointable_Engine<Engine>
{
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  using coord_t = typename Lattice::coord_type;
  static_assert(std::is_trivially_copyable_v<state_t>,
                "snapshot_frame_bytes requires a trivially copyable state_type");
  static_assert(Lattice::rank <= 3,
                "snapshot_frame_bytes currently supports rank <= 3");
 
  const Lattice &frame = engine.frame();
  std::uint64_t cell_count = 1;
  for (std::size_t d = 0; d < Lattice::rank; ++d)
    cell_count *= static_cast<std::uint64_t>(frame.size(d));
 
  std::vector<std::uint8_t> bytes(static_cast<std::size_t>(cell_count) * sizeof(state_t));
  std::size_t off = 0;
  auto append = [&](const state_t &v)
  {
    std::memcpy(bytes.data() + off, &v, sizeof(state_t));
    off += sizeof(state_t);
  };
 
  if constexpr (Lattice::rank == 1)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        append(frame.at(coord_t{static_cast<ca_index_t>(i)}));
    }
  else if constexpr (Lattice::rank == 2)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        for (ca_size_t j = 0; j < frame.size(1); ++j)
          append(frame.at(coord_t{static_cast<ca_index_t>(i),
                                  static_cast<ca_index_t>(j)}));
    }
  else if constexpr (Lattice::rank == 3)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        for (ca_size_t j = 0; j < frame.size(1); ++j)
          for (ca_size_t k = 0; k < frame.size(2); ++k)
            append(frame.at(coord_t{static_cast<ca_index_t>(i),
                                    static_cast<ca_index_t>(j),
                                    static_cast<ca_index_t>(k)}));
    }
  return bytes;
}
 
template <typename Engine>
void restore_frame_from_bytes(Engine &engine, const std::vector<std::uint8_t> &bytes)
  requires Checkpointable_Engine<Engine>
{
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  using coord_t = typename Lattice::coord_type;
  static_assert(std::is_trivially_copyable_v<state_t>,
                "restore_frame_from_bytes requires a trivially copyable state_type");
  static_assert(Lattice::rank <= 3,
                "restore_frame_from_bytes currently supports rank <= 3");
 
  const Lattice &frame = engine.frame();
  std::uint64_t cell_count = 1;
  for (std::size_t d = 0; d < Lattice::rank; ++d)
    cell_count *= static_cast<std::uint64_t>(frame.size(d));
  ah_runtime_error_if(bytes.size() != static_cast<std::size_t>(cell_count) * sizeof(state_t))
    << "restore_frame_from_bytes: size mismatch (expected "
    << (cell_count * sizeof(state_t)) << ", got " << bytes.size() << ")";
 
  std::size_t off = 0;
  auto next = [&]() -> state_t
  {
    state_t v{};
    std::memcpy(&v, bytes.data() + off, sizeof(state_t));
    off += sizeof(state_t);
    return v;
  };
 
  if constexpr (Lattice::rank == 1)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        engine.write_lattice().set(coord_t{static_cast<ca_index_t>(i)}, next());
    }
  else if constexpr (Lattice::rank == 2)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        for (ca_size_t j = 0; j < frame.size(1); ++j)
          engine.write_lattice().set(
            coord_t{static_cast<ca_index_t>(i), static_cast<ca_index_t>(j)},
            next());
    }
  else if constexpr (Lattice::rank == 3)
    {
      for (ca_size_t i = 0; i < frame.size(0); ++i)
        for (ca_size_t j = 0; j < frame.size(1); ++j)
          for (ca_size_t k = 0; k < frame.size(2); ++k)
            engine.write_lattice().set(coord_t{static_cast<ca_index_t>(i),
                                               static_cast<ca_index_t>(j),
                                               static_cast<ca_index_t>(k)},
                                       next());
    }
}
 
struct Header_Snapshot
{
  std::uint64_t engine_type_hash = 0;
  std::uint32_t state_type_size = 0;
  std::uint32_t rank = 0;
  std::array<std::uint64_t, 4> extents{};
  std::uint64_t step_count = 0;
  std::uint8_t has_rng = 0;
  std::uint64_t master_seed = 0;
  std::uint64_t cell_count = 0;
};
 
template <typename Engine>
[[nodiscard]] Header_Snapshot capture_header(const Engine &engine)
  requires Checkpointable_Engine<Engine>
{
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  Header_Snapshot hs;
  hs.engine_type_hash = type_hash<Engine>();
  hs.state_type_size = static_cast<std::uint32_t>(sizeof(state_t));
  hs.rank = static_cast<std::uint32_t>(Lattice::rank);
  const auto &ext = engine.frame().extents();
  for (std::size_t d = 0; d < Lattice::rank; ++d)
    hs.extents[d] = static_cast<std::uint64_t>(ext[d]);
  hs.step_count = static_cast<std::uint64_t>(engine.steps_run());
  std::uint64_t n = 1;
  for (std::size_t d = 0; d < Lattice::rank; ++d)
    n *= hs.extents[d];
  hs.cell_count = n;
  if constexpr (Has_Master_Seed<typename Engine::rule_type>)
    {
      hs.has_rng = 1;
      hs.master_seed = static_cast<std::uint64_t>(engine.rule().master_seed());
    }
  return hs;
}
 
inline void write_header_bytes(std::ostream &out,
                               const Header_Snapshot &hs,
                               const std::uint32_t flags,
                               const std::uint32_t compression_level,
                               const std::uint64_t payload_size,
                               const std::uint64_t delta_base_step)
{
  out.write(magic_bytes.data(), magic_bytes.size());
  write_le<std::uint32_t>(out, format_version);
  write_le<std::uint64_t>(out, hs.engine_type_hash);
  write_le<std::uint32_t>(out, hs.state_type_size);
  write_le<std::uint32_t>(out, hs.rank);
  for (std::size_t d = 0; d < hs.extents.size(); ++d)
    write_le<std::uint64_t>(out, hs.extents[d]);
  write_le<std::uint64_t>(out, hs.step_count);
  write_le<std::uint8_t>(out, hs.has_rng);
  for (int i = 0; i < 7; ++i)
    write_le<std::uint8_t>(out, std::uint8_t{0});
  write_le<std::uint64_t>(out, hs.master_seed);
  write_le<std::uint64_t>(out, hs.cell_count);
  write_le<std::uint32_t>(out, flags);
  write_le<std::uint32_t>(out, compression_level);
  write_le<std::uint64_t>(out, payload_size);
  write_le<std::uint64_t>(out, delta_base_step);
}
 
inline void atomic_write_file(const std::filesystem::path &path,
                              const Header_Snapshot &hs,
                              const std::uint32_t flags,
                              const std::uint32_t compression_level,
                              const std::vector<std::uint8_t> &payload,
                              const std::uint64_t delta_base_step,
                              const Checkpoint_Options &options)
{
  const std::filesystem::path tmp = make_tmp_path(path);
  Tmp_File_Guard guard(tmp);
  {
    std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
    ah_runtime_error_if(not out)
      << "atomic_write_file: cannot open '" << tmp.string() << "' for write";
    write_header_bytes(out,
                       hs,
                       flags,
                       compression_level,
                       static_cast<std::uint64_t>(payload.size()),
                       delta_base_step);
    if (not payload.empty())
      out.write(reinterpret_cast<const char *>(payload.data()),
                static_cast<std::streamsize>(payload.size()));
    out.flush();
    ah_runtime_error_if(not out)
      << "atomic_write_file: write failed for '" << tmp.string() << "'";
  }
  sync_file(tmp);
 
  std::error_code ec;
  std::filesystem::rename(tmp, path, ec);
  ah_runtime_error_if(ec)
    << "atomic_write_file: rename '" << tmp.string() << "' -> '"
    << path.string() << "' failed: " << ec.message();
  guard.commit();
 
  if (options.sync_dir)
    {
      std::filesystem::path parent = path.parent_path();
      if (parent.empty())
        parent = std::filesystem::path(".");
      sync_directory(parent);
    }
}
 
inline std::vector<std::uint8_t>
build_payload(const std::vector<std::uint8_t> &raw,
              const Checkpoint_Options &options,
              std::uint32_t &flags_out,
              std::uint32_t &level_out)
{
  if (not options.compress or raw.empty())
    {
      flags_out = 0;
      level_out = 0;
      return raw;
    }
  const int level = (options.level <= 0) ? 6 : options.level;
  std::vector<std::uint8_t> compressed = deflate_bytes(raw.data(), raw.size(), level);
  flags_out = ca_checkpoint_flags::compressed;
  level_out = static_cast<std::uint32_t>(level);
  return compressed;
}
 
inline std::vector<std::uint8_t>
read_raw_payload(const std::filesystem::path &path, const Checkpoint_Header &h)
{
  std::ifstream in(path, std::ios::binary);
  ah_runtime_error_if(not in)
    << "read_raw_payload: cannot reopen '" << path.string() << "'";
  const std::streamoff header_bytes
    = (h.format_version >= 2) ? header_bytes_v2 : header_bytes_v1;
 
  // Hardening (CWE-789): the on-disk payload occupies exactly the bytes
  // after the header. A hostile header that claims a payload larger than
  // the file would otherwise drive a multi-gigabyte allocation below.
  const std::uint64_t file_size
    = static_cast<std::uint64_t>(std::filesystem::file_size(path));
  const std::uint64_t hdr = static_cast<std::uint64_t>(header_bytes);
  ah_runtime_error_if(file_size < hdr or h.payload_size > file_size - hdr)
    << "read_raw_payload: declared payload_size " << h.payload_size
    << " exceeds the file body ("
    << (file_size >= hdr ? file_size - hdr : 0)
    << " bytes) in '" << path.string() << "'";
 
  in.seekg(header_bytes, std::ios::beg);
  std::vector<std::uint8_t> payload(static_cast<std::size_t>(h.payload_size));
  if (not payload.empty())
    {
      in.read(reinterpret_cast<char *>(payload.data()),
              static_cast<std::streamsize>(payload.size()));
      ah_runtime_error_if(not in)
        << "read_raw_payload: short read of payload in '" << path.string() << "'";
    }
  if (h.flags & ca_checkpoint_flags::compressed)
    {
      const std::size_t expected
        = static_cast<std::size_t>(h.cell_count) * h.state_type_size;
      return inflate_bytes(payload.data(), payload.size(), expected);
    }
  return payload;
}
 
}  // namespace ca_checkpoint_detail
 
// =====================================================================
// save_checkpoint
// =====================================================================
 
template <typename Engine>
void save_checkpoint(Engine &engine,
                     const std::filesystem::path &path,
                     const Checkpoint_Options &options = {})
  requires Checkpointable_Engine<Engine>
{
  using namespace ca_checkpoint_detail;
  const Header_Snapshot hs = capture_header(engine);
  std::vector<std::uint8_t> raw = snapshot_frame_bytes(engine);
  std::uint32_t flags = 0;
  std::uint32_t level = 0;
  std::vector<std::uint8_t> payload = build_payload(raw, options, flags, level);
  atomic_write_file(path, hs, flags, level, payload, /*delta_base_step=*/0, options);
}
 
// =====================================================================
// load_checkpoint_into
// =====================================================================
 
template <typename Engine>
[[nodiscard]] Resume_Token load_checkpoint_into(Engine &engine,
                                                const std::filesystem::path &path)
  requires Checkpointable_Engine<Engine>
{
  using namespace ca_checkpoint_detail;
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  static_assert(std::is_trivially_copyable_v<state_t>,
                "load_checkpoint_into requires a trivially copyable state_type");
 
  Resume_Token token;
  token.source_path = path;
  token.header = inspect_checkpoint(path);
 
  ah_runtime_error_if(token.header.engine_type_hash != type_hash<Engine>())
    << "load_checkpoint_into: type hash mismatch in '" << path.string()
    << "' (file=" << token.header.engine_type_hash
    << " engine=" << type_hash<Engine>() << ")";
  ah_runtime_error_if(token.header.state_type_size != sizeof(state_t))
    << "load_checkpoint_into: state size mismatch in '" << path.string()
    << "' (file=" << token.header.state_type_size << " engine=" << sizeof(state_t) << ")";
  ah_runtime_error_if(token.header.rank != Lattice::rank)
    << "load_checkpoint_into: rank mismatch in '" << path.string()
    << "' (file=" << token.header.rank << " engine=" << Lattice::rank << ")";
  ah_runtime_error_if(token.header.flags & ca_checkpoint_flags::delta)
    << "load_checkpoint_into: '" << path.string() << "' is a delta snapshot; "
    << "use apply_delta_checkpoint() after loading the baseline";
 
  const auto &ext = engine.frame().extents();
  for (std::size_t d = 0; d < Lattice::rank; ++d)
    ah_runtime_error_if(token.header.extents[d] != static_cast<std::uint64_t>(ext[d]))
      << "load_checkpoint_into: extent[" << d << "] mismatch in '" << path.string()
      << "' (file=" << token.header.extents[d] << " engine=" << ext[d] << ")";
 
  std::vector<std::uint8_t> raw = read_raw_payload(path, token.header);
  restore_frame_from_bytes(engine, raw);
 
  // Restore step counter.
  engine.set_step_count(token.header.step_count);
 
  // Restore master RNG seed when both sides agree on the contract.
  if constexpr (Has_Master_Seed<typename Engine::rule_type>)
    {
      if (token.header.has_rng != 0)
        engine.rule().set_master_seed(token.header.master_seed);
    }
  return token;
}
 
// =====================================================================
// Delta snapshots (best-effort)
// =====================================================================
 
template <typename Engine>
void save_delta_checkpoint(Engine &engine,
                           const std::vector<std::uint8_t> &baseline,
                           const std::uint64_t base_step,
                           const std::filesystem::path &path,
                           const Checkpoint_Options &options = {})
  requires Checkpointable_Engine<Engine>
{
  using namespace ca_checkpoint_detail;
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  static_assert(std::is_trivially_copyable_v<state_t>,
                "save_delta_checkpoint requires a trivially copyable state_type");
 
  const Header_Snapshot hs = capture_header(engine);
  const std::size_t cell_bytes = sizeof(state_t);
  const std::size_t total_cells = static_cast<std::size_t>(hs.cell_count);
  ah_runtime_error_if(baseline.size() != total_cells * cell_bytes)
    << "save_delta_checkpoint: baseline size mismatch (expected "
    << (total_cells * cell_bytes) << ", got " << baseline.size() << ")";
 
  std::vector<std::uint8_t> current = snapshot_frame_bytes(engine);
  std::vector<std::uint8_t> diff;
  diff.reserve(64);  // arbitrary small reserve to avoid tiny reallocs.
  for (std::size_t i = 0; i < total_cells; ++i)
    {
      const std::size_t off = i * cell_bytes;
      if (std::memcmp(current.data() + off,
                      baseline.data() + off,
                      cell_bytes)
          != 0)
        {
          const std::uint64_t idx = static_cast<std::uint64_t>(i);
          const std::size_t head = diff.size();
          diff.resize(head + sizeof(std::uint64_t) + cell_bytes);
          std::memcpy(diff.data() + head, &idx, sizeof(std::uint64_t));
          std::memcpy(diff.data() + head + sizeof(std::uint64_t),
                      current.data() + off,
                      cell_bytes);
        }
    }
 
  std::uint32_t flags = ca_checkpoint_flags::delta;
  std::uint32_t level = 0;
  std::vector<std::uint8_t> payload;
  if (options.compress and not diff.empty())
    {
      const int lvl = (options.level <= 0) ? 6 : options.level;
      payload = deflate_bytes(diff.data(), diff.size(), lvl);
      flags |= ca_checkpoint_flags::compressed;
      level = static_cast<std::uint32_t>(lvl);
    }
  else
    payload = std::move(diff);
 
  atomic_write_file(path, hs, flags, level, payload, base_step, options);
}
 
template <typename Engine>
[[nodiscard]] Resume_Token
apply_delta_checkpoint(Engine &engine, const std::filesystem::path &path)
  requires Checkpointable_Engine<Engine>
{
  using namespace ca_checkpoint_detail;
  using Lattice = typename Engine::lattice_type;
  using state_t = typename Lattice::state_type;
  using coord_t = typename Lattice::coord_type;
  static_assert(std::is_trivially_copyable_v<state_t>,
                "apply_delta_checkpoint requires a trivially copyable state_type");
  static_assert(Lattice::rank <= 3,
                "apply_delta_checkpoint currently supports rank <= 3");
 
  Resume_Token token;
  token.source_path = path;
  token.header = inspect_checkpoint(path);
 
  ah_runtime_error_if(not (token.header.flags & ca_checkpoint_flags::delta))
    << "apply_delta_checkpoint: '" << path.string() << "' is not a delta snapshot";
  ah_runtime_error_if(token.header.engine_type_hash != type_hash<Engine>())
    << "apply_delta_checkpoint: type hash mismatch in '" << path.string() << "'";
  ah_runtime_error_if(token.header.state_type_size != sizeof(state_t))
    << "apply_delta_checkpoint: state size mismatch in '" << path.string() << "'";
  ah_runtime_error_if(token.header.rank != Lattice::rank)
    << "apply_delta_checkpoint: rank mismatch in '" << path.string() << "'";
 
  std::ifstream in(path, std::ios::binary);
  ah_runtime_error_if(not in)
    << "apply_delta_checkpoint: cannot reopen '" << path.string() << "'";
  in.seekg(header_bytes_v2, std::ios::beg);
  std::vector<std::uint8_t> payload(static_cast<std::size_t>(token.header.payload_size));
  if (not payload.empty())
    {
      in.read(reinterpret_cast<char *>(payload.data()),
              static_cast<std::streamsize>(payload.size()));
      ah_runtime_error_if(not in)
        << "apply_delta_checkpoint: short read in '" << path.string() << "'";
    }
 
  std::vector<std::uint8_t> diff;
  if (token.header.flags & ca_checkpoint_flags::compressed)
    {
      // Upper bound on the uncompressed delta size: at most every cell
      // is mutated, producing `cell_count * (8 + state_size)` bytes.
      const std::size_t bound
        = static_cast<std::size_t>(token.header.cell_count)
          * (sizeof(std::uint64_t) + sizeof(state_t));
      diff = inflate_bytes_up_to(payload.data(), payload.size(), bound);
    }
  else
    diff = std::move(payload);
 
  const std::size_t cell_bytes = sizeof(state_t);
  const std::size_t entry_bytes = sizeof(std::uint64_t) + cell_bytes;
  ah_runtime_error_if(diff.size() % entry_bytes != 0)
    << "apply_delta_checkpoint: delta payload size " << diff.size()
    << " is not a multiple of (" << entry_bytes << ")";
 
  auto extents_from_header = [&](std::size_t d) -> ca_size_t
  { return static_cast<ca_size_t>(token.header.extents[d]); };
 
  const std::size_t entries = diff.size() / entry_bytes;
  for (std::size_t e = 0; e < entries; ++e)
    {
      std::uint64_t idx = 0;
      state_t value{};
      std::memcpy(&idx, diff.data() + e * entry_bytes, sizeof(std::uint64_t));
      std::memcpy(&value,
                  diff.data() + e * entry_bytes + sizeof(std::uint64_t),
                  cell_bytes);
      ah_runtime_error_if(idx >= token.header.cell_count)
        << "apply_delta_checkpoint: linear index " << idx
        << " out of range (cell_count=" << token.header.cell_count << ")";
 
      if constexpr (Lattice::rank == 1)
        {
          engine.write_lattice().set(
            coord_t{static_cast<ca_index_t>(idx)},
            value);
        }
      else if constexpr (Lattice::rank == 2)
        {
          const auto j = static_cast<ca_index_t>(idx % extents_from_header(1));
          const auto i = static_cast<ca_index_t>(idx / extents_from_header(1));
          engine.write_lattice().set(coord_t{i, j}, value);
        }
      else if constexpr (Lattice::rank == 3)
        {
          const auto e2 = extents_from_header(2);
          const auto e1 = extents_from_header(1);
          const auto k = static_cast<ca_index_t>(idx % e2);
          const auto j = static_cast<ca_index_t>((idx / e2) % e1);
          const auto i = static_cast<ca_index_t>(idx / (e1 * e2));
          engine.write_lattice().set(coord_t{i, j, k}, value);
        }
    }
 
  engine.set_step_count(token.header.step_count);
  if constexpr (Has_Master_Seed<typename Engine::rule_type>)
    {
      if (token.header.has_rng != 0)
        engine.rule().set_master_seed(token.header.master_seed);
    }
  return token;
}
 
// =====================================================================
// Async_Checkpoint_Writer
// =====================================================================
 
template <typename Engine>
class Async_Checkpoint_Writer
{
public:
  enum class Queue_Policy
  {
    Block,
    Drop_Oldest
  };
 
  explicit Async_Checkpoint_Writer(const std::size_t capacity = 8,
                                   const Queue_Policy policy = Queue_Policy::Block)
    : capacity_(capacity)
    , policy_(policy)
  {
    ah_domain_error_if(capacity == 0)
      << "Async_Checkpoint_Writer: capacity must be >= 1";
    worker_ = std::thread([this] { this->worker_loop(); });
  }
 
  Async_Checkpoint_Writer(const Async_Checkpoint_Writer &) = delete;
  Async_Checkpoint_Writer &operator=(const Async_Checkpoint_Writer &) = delete;
 
  ~Async_Checkpoint_Writer() noexcept
  {
    try
      {
        flush();
      }
    catch (...)
      {
        // Swallow flush errors during destruction; the worker thread
        // is still joined below.
      }
    {
      std::unique_lock<std::mutex> lk(mu_);
      stop_ = true;
      not_empty_.notify_all();
    }
    if (worker_.joinable())
      worker_.join();
  }
 
  void submit(Engine &engine,
              std::filesystem::path path,
              Checkpoint_Options options = {})
  {
    Task task;
    task.header = ca_checkpoint_detail::capture_header(engine);
    task.raw = ca_checkpoint_detail::snapshot_frame_bytes(engine);
    task.path = std::move(path);
    task.options = options;
 
    std::unique_lock<std::mutex> lk(mu_);
    if (policy_ == Queue_Policy::Drop_Oldest)
      {
        while (queue_.size() >= capacity_)
          {
            queue_.pop_front();
            total_dropped_.fetch_add(1, std::memory_order_relaxed);
          }
      }
    else
      {
        not_full_.wait(lk, [this] { return queue_.size() < capacity_ or stop_; });
      }
    queue_.push_back(std::move(task));
    not_empty_.notify_one();
  }
 
  void flush()
  {
    std::unique_lock<std::mutex> lk(mu_);
    drained_.wait(lk, [this] { return queue_.empty() and not in_flight_; });
    if (worker_error_)
      {
        const std::string msg = std::move(*worker_error_);
        worker_error_.reset();
        ah_runtime_error_if(true) << msg;
      }
  }
 
  [[nodiscard]] std::size_t pending() const
  {
    std::lock_guard<std::mutex> lk(mu_);
    return queue_.size();
  }
 
  void set_before_write_hook(std::function<void()> hook)
  {
    before_write_hook_ = std::move(hook);
  }
 
  [[nodiscard]] std::size_t total_written() const noexcept
  {
    return total_written_.load(std::memory_order_relaxed);
  }
 
  [[nodiscard]] std::size_t total_dropped() const noexcept
  {
    return total_dropped_.load(std::memory_order_relaxed);
  }
 
  [[nodiscard]] std::size_t capacity() const noexcept { return capacity_; }
 
  [[nodiscard]] Queue_Policy policy() const noexcept { return policy_; }
 
private:
  struct Task
  {
    ca_checkpoint_detail::Header_Snapshot header;
    std::vector<std::uint8_t> raw;
    std::filesystem::path path;
    Checkpoint_Options options;
  };
 
  void worker_loop() noexcept
  {
    for (;;)
      {
        Task task;
        {
          std::unique_lock<std::mutex> lk(mu_);
          not_empty_.wait(lk, [this] { return not queue_.empty() or stop_; });
          if (queue_.empty() and stop_)
            return;
          task = std::move(queue_.front());
          queue_.pop_front();
          in_flight_ = true;
          not_full_.notify_one();
        }
 
        if (before_write_hook_)
          before_write_hook_();
 
        bool ok = true;
        std::string err;
        try
          {
            std::uint32_t flags = 0;
            std::uint32_t level = 0;
            std::vector<std::uint8_t> payload
              = ca_checkpoint_detail::build_payload(task.raw, task.options, flags, level);
            ca_checkpoint_detail::atomic_write_file(task.path,
                                                    task.header,
                                                    flags,
                                                    level,
                                                    payload,
                                                    /*delta_base_step=*/0,
                                                    task.options);
          }
        catch (const std::exception &e)
          {
            ok = false;
            err = e.what();
          }
        catch (...)
          {
            ok = false;
            err = "Async_Checkpoint_Writer: unknown exception";
          }
 
        {
          std::unique_lock<std::mutex> lk(mu_);
          in_flight_ = false;
          if (ok)
            total_written_.fetch_add(1, std::memory_order_relaxed);
          else if (not worker_error_)
            worker_error_ = std::move(err);
          if (queue_.empty())
            drained_.notify_all();
        }
      }
  }
 
  mutable std::mutex mu_;
  std::condition_variable not_empty_;
  std::condition_variable not_full_;
  std::condition_variable drained_;
  std::deque<Task> queue_;
  std::size_t capacity_;
  Queue_Policy policy_;
  bool stop_ = false;
  bool in_flight_ = false;
  std::optional<std::string> worker_error_;
  std::atomic<std::size_t> total_written_{0};
  std::atomic<std::size_t> total_dropped_{0};
  std::function<void()> before_write_hook_;
  std::thread worker_;
};
 
// =====================================================================
// Periodic_Checkpoint_Observer
// =====================================================================
 
template <typename Engine>
class Periodic_Checkpoint_Observer
{
  Engine *engine_;
  std::size_t every_;
  std::string pattern_;
  std::size_t zero_pad_;
  std::filesystem::path last_path_;
  Async_Checkpoint_Writer<Engine> *async_writer_ = nullptr;
  Checkpoint_Options options_{};
 
  [[nodiscard]] std::filesystem::path format_path(const std::size_t step) const
  {
    constexpr std::string_view placeholder = "{step}";
    const std::size_t pos = pattern_.find(placeholder);
    std::string padded = std::to_string(step);
    if (padded.size() < zero_pad_)
      padded.insert(0, zero_pad_ - padded.size(), '0');
    if (pos == std::string::npos)
      return std::filesystem::path(pattern_ + "." + padded);
    return std::filesystem::path(pattern_.substr(0, pos) + padded
                                 + pattern_.substr(pos + placeholder.size()));
  }
 
public:
  Periodic_Checkpoint_Observer(Engine &engine,
                               const std::size_t every,
                               std::string path_pattern,
                               const std::size_t zero_pad = 6)
    : engine_(&engine), every_(every), pattern_(std::move(path_pattern)), zero_pad_(zero_pad)
  {
    ah_domain_error_if(every == 0)
      << "Periodic_Checkpoint_Observer: every must be >= 1";
  }
 
  Periodic_Checkpoint_Observer(Engine &engine,
                               const std::size_t every,
                               std::string path_pattern,
                               Async_Checkpoint_Writer<Engine> &writer,
                               Checkpoint_Options options = {},
                               const std::size_t zero_pad = 6)
    : engine_(&engine)
    , every_(every)
    , pattern_(std::move(path_pattern))
    , zero_pad_(zero_pad)
    , async_writer_(&writer)
    , options_(options)
  {
    ah_domain_error_if(every == 0)
      << "Periodic_Checkpoint_Observer: every must be >= 1";
  }
 
  template <typename Lattice>
  void on_step_begin(const std::size_t step, const Lattice &frame) noexcept
  {
    (void) step;
    (void) frame;
  }
 
  template <typename Lattice>
  void on_step_end(const std::size_t step, const Lattice &)
  {
    if (step == 0 or (step % every_) != 0)
      return;
    last_path_ = format_path(step);
    if (async_writer_ != nullptr)
      async_writer_->submit(*engine_, last_path_, options_);
    else
      save_checkpoint(*engine_, last_path_, options_);
  }
 
  [[nodiscard]] const std::filesystem::path &last_path() const noexcept
  {
    return last_path_;
  }
};
 
}  // namespace CA
}  // namespace Aleph
 
#endif  // CA_CHECKPOINT_H