raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CC1: persistent file-backed spool

Browse Source 
Adds opt-in disk persistence to SpoolStore.  `enable_persistence(dir)`
turns every enqueue into a single `<seq>.spool` file alongside the
in-memory queue; drain and clear delete the matching files; restart
replays the directory sorted by seq.

Writes are atomic: serialize the message via the SECS-II codec, write
to `.tmp`, then `std::filesystem::rename` to the final name.  Malformed
records are dropped silently so a single bad file can't poison the
whole spool.

`secs_server --spool-dir <path>` enables persistence at startup.
Without the flag the behaviour is identical to before (in-memory only).

Two new tests: enqueue → restart → replay → drain restores the wire
order, and clear deletes the journal files.

Test suite: 291 cases / 1515 assertions.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-08 23:51:46 +02:00
parent cfa2d1e531
commit d69f26b415
3 changed files with 250 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/secsgem/gem/store/spool.hpp
+167 -5
View File
@@ -1,11 +1,20 @@
#pragma once
#include <algorithm>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <deque>
#include <filesystem>
#include <fstream>
#include <optional>
#include <set>
#include <string>
#include <system_error>
#include <utility>
#include <vector>
#include "secsgem/secs2/codec.hpp"
#include "secsgem/secs2/message.hpp"
// E30 §6.22 Spooling.
@@ -20,6 +29,13 @@
// S2F43 (Reset Spooling). This implementation supports the simple form
// (per-stream enable list, no per-message-type granularity) which covers
// the common cases (stream 5 alarms + stream 6 event reports).
//
// Persistence (opt-in): when `enable_persistence(dir)` is called, each
// enqueued message is also written as a single file `<seq>.spool` in
// the configured directory; on restart, `enable_persistence` replays
// the directory back into the in-memory queue. Drain and clear delete
// the corresponding files. Writes are atomic: write to `.tmp`, fsync,
// rename.
namespace secsgem::gem {
@@ -76,36 +92,182 @@ class SpoolStore {
void set_force_spool(bool on) { force_spool_ = on; }
bool force_spool() const { return force_spool_; }
// ---- Persistence ----------------------------------------------------
// Enable file-backed spooling at `dir`. Creates the directory if it
// doesn't exist; scans existing `*.spool` files and replays them
// back into the queue (sorted by seq) before returning. Subsequent
// enqueue / drain / clear calls keep the directory in sync. Errors
// (permission, malformed records) are silently skipped per file so a
// single bad record can't poison the whole spool — the worst case is
// dropping that one message.
void enable_persistence(std::filesystem::path dir) {
namespace fs = std::filesystem;
journal_dir_ = std::move(dir);
persistent_ = true;
std::error_code ec;
fs::create_directories(journal_dir_, ec);
// Replay: collect all .spool files, sort by integer seq prefix.
std::vector<std::pair<uint64_t, fs::path>> existing;
for (auto& entry : fs::directory_iterator(journal_dir_, ec)) {
if (!entry.is_regular_file()) continue;
const auto& p = entry.path();
if (p.extension() != ".spool") continue;
uint64_t seq = 0;
try { seq = std::stoull(p.stem().string()); }
catch (...) { continue; }
existing.emplace_back(seq, p);
}
std::sort(existing.begin(), existing.end(),
[](const auto& a, const auto& b) { return a.first < b.first; });
for (auto& [seq, path] : existing) {
auto msg = load_record(path);
if (!msg) {
// unreadable / malformed; drop the file so it doesn't linger.
fs::remove(path, ec);
continue;
}
queue_.push_back({std::move(*msg), path});
if (seq >= next_seq_) next_seq_ = seq + 1;
}
}
bool persistence_enabled() const { return persistent_; }
std::filesystem::path journal_dir() const { return journal_dir_; }
EnqueueResult enqueue(s2::Message msg) {
if (!is_stream_spoolable(msg.stream)) return EnqueueResult::Dropped_NotSpoolable;
if (queue_.size() >= max_size_) {
// Overflow policy: drop the oldest (FIFO eviction).
if (persistent_ && !queue_.front().path.empty()) {
std::error_code ec;
std::filesystem::remove(queue_.front().path, ec);
}
queue_.pop_front();
}
queue_.push_back(std::move(msg));
std::filesystem::path path;
if (persistent_) {
path = write_record(msg);
}
queue_.push_back({std::move(msg), std::move(path)});
return EnqueueResult::Queued;
}
// Drain everything currently in the queue (FIFO order).
// Drain everything currently in the queue (FIFO order). Also deletes
// the matching journal files when persistence is enabled.
std::vector<s2::Message> drain() {
std::vector<s2::Message> out;
out.reserve(queue_.size());
std::error_code ec;
while (!queue_.empty()) {
out.push_back(std::move(queue_.front()));
auto& slot = queue_.front();
out.push_back(std::move(slot.msg));
if (persistent_ && !slot.path.empty()) {
std::filesystem::remove(slot.path, ec);
}
queue_.pop_front();
}
return out;
}
void clear() { queue_.clear(); }
void clear() {
if (persistent_) {
std::error_code ec;
for (auto& slot : queue_) {
if (!slot.path.empty()) std::filesystem::remove(slot.path, ec);
}
}
queue_.clear();
}
std::size_t size() const { return queue_.size(); }
bool empty() const { return queue_.empty(); }
private:
struct Slot {
s2::Message msg;
std::filesystem::path path; // empty when persistence is disabled
};
// Journal record layout (little-endian-free; all-bytes are explicit):
// [u8 magic = 0xE5]
// [u8 version = 1]
// [u8 stream]
// [u8 function]
// [u8 reply_expected]
// [u32 body_length, big-endian]
// [body_length bytes]
static constexpr uint8_t kMagic = 0xE5;
static constexpr uint8_t kVersion = 0x01;
std::filesystem::path write_record(const s2::Message& msg) {
namespace fs = std::filesystem;
auto seq = next_seq_++;
char namebuf[32];
std::snprintf(namebuf, sizeof(namebuf), "%010llu.spool",
static_cast<unsigned long long>(seq));
fs::path final_path = journal_dir_ / namebuf;
fs::path tmp_path = final_path;
tmp_path += ".tmp";
auto body_bytes = msg.body ? secs2::encode(*msg.body) : std::vector<uint8_t>{};
const uint32_t blen = static_cast<uint32_t>(body_bytes.size());
std::ofstream out(tmp_path, std::ios::binary | std::ios::trunc);
if (!out) return {};
uint8_t hdr[5] = {kMagic, kVersion, msg.stream, msg.function,
static_cast<uint8_t>(msg.reply_expected ? 1 : 0)};
out.write(reinterpret_cast<const char*>(hdr), sizeof(hdr));
uint8_t blen_be[4] = {
static_cast<uint8_t>((blen >> 24) & 0xFF),
static_cast<uint8_t>((blen >> 16) & 0xFF),
static_cast<uint8_t>((blen >> 8) & 0xFF),
static_cast<uint8_t>(blen & 0xFF)};
out.write(reinterpret_cast<const char*>(blen_be), sizeof(blen_be));
if (blen) out.write(reinterpret_cast<const char*>(body_bytes.data()), blen);
out.flush();
out.close();
std::error_code ec;
fs::rename(tmp_path, final_path, ec);
if (ec) {
fs::remove(tmp_path, ec);
return {};
}
return final_path;
}
static std::optional<s2::Message> load_record(const std::filesystem::path& p) {
std::ifstream in(p, std::ios::binary);
if (!in) return std::nullopt;
uint8_t hdr[5];
in.read(reinterpret_cast<char*>(hdr), sizeof(hdr));
if (!in || hdr[0] != kMagic || hdr[1] != kVersion) return std::nullopt;
uint8_t blen_be[4];
in.read(reinterpret_cast<char*>(blen_be), sizeof(blen_be));
if (!in) return std::nullopt;
const uint32_t blen = (uint32_t{blen_be[0]} << 24) |
(uint32_t{blen_be[1]} << 16) |
(uint32_t{blen_be[2]} << 8) |
uint32_t{blen_be[3]};
std::vector<uint8_t> body(blen);
if (blen) {
in.read(reinterpret_cast<char*>(body.data()), blen);
if (!in) return std::nullopt;
}
try {
return s2::Message::from_body(hdr[2], hdr[3], hdr[4] != 0, body);
} catch (...) {
return std::nullopt;
}
}
std::set<uint8_t> spoolable_;
std::deque<s2::Message> queue_;
std::deque<Slot> queue_;
std::size_t max_size_ = 100;
bool force_spool_ = false;
bool persistent_ = false;
std::filesystem::path journal_dir_;
uint64_t next_seq_ = 0;
};
} // namespace secsgem::gem