From d69f26b41510fd6b11450c22d51b723ef053910b Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Mon, 8 Jun 2026 23:51:46 +0200 Subject: [PATCH] CC1: persistent file-backed spool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds opt-in disk persistence to SpoolStore. `enable_persistence(dir)` turns every enqueue into a single `.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 ` 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 --- apps/secs_server.cpp | 6 + include/secsgem/gem/store/spool.hpp | 172 +++++++++++++++++++++++++++- tests/test_data_model.cpp | 77 +++++++++++++ 3 files changed, 250 insertions(+), 5 deletions(-) diff --git a/apps/secs_server.cpp b/apps/secs_server.cpp index 8ecba49..c6d3a6d 100644 --- a/apps/secs_server.cpp +++ b/apps/secs_server.cpp @@ -49,10 +49,16 @@ int main(int argc, char** argv) { const auto port = static_cast(std::stoi(arg(argc, argv, "--port", "5000"))); const auto equipment_yaml = arg(argc, argv, "--config", "/app/data/equipment.yaml"); const auto state_yaml = arg(argc, argv, "--state-table", "/app/data/control_state.yaml"); + const auto spool_dir = arg(argc, argv, "--spool-dir", ""); auto logfn = [](const std::string& m) { std::cout << "[equip] " << m << std::endl; }; auto model = std::make_shared(); + if (!spool_dir.empty()) { + model->spool.enable_persistence(spool_dir); + logfn("spool: persisting to " + spool_dir + + " (replayed " + std::to_string(model->spool.size()) + " messages)"); + } config::EquipmentDescriptor desc; config::ControlStateConfig sm_cfg; try { diff --git a/include/secsgem/gem/store/spool.hpp b/include/secsgem/gem/store/spool.hpp index f2d9fdb..d2d336f 100644 --- a/include/secsgem/gem/store/spool.hpp +++ b/include/secsgem/gem/store/spool.hpp @@ -1,11 +1,20 @@ #pragma once +#include #include +#include +#include #include +#include +#include +#include #include +#include +#include #include #include +#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 `.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> 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 drain() { std::vector 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(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{}; + const uint32_t blen = static_cast(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(msg.reply_expected ? 1 : 0)}; + out.write(reinterpret_cast(hdr), sizeof(hdr)); + uint8_t blen_be[4] = { + static_cast((blen >> 24) & 0xFF), + static_cast((blen >> 16) & 0xFF), + static_cast((blen >> 8) & 0xFF), + static_cast(blen & 0xFF)}; + out.write(reinterpret_cast(blen_be), sizeof(blen_be)); + if (blen) out.write(reinterpret_cast(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 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(hdr), sizeof(hdr)); + if (!in || hdr[0] != kMagic || hdr[1] != kVersion) return std::nullopt; + uint8_t blen_be[4]; + in.read(reinterpret_cast(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 body(blen); + if (blen) { + in.read(reinterpret_cast(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 spoolable_; - std::deque queue_; + std::deque 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 diff --git a/tests/test_data_model.cpp b/tests/test_data_model.cpp index ba9f0ac..7d72b81 100644 --- a/tests/test_data_model.cpp +++ b/tests/test_data_model.cpp @@ -1,6 +1,10 @@ #include +#include #include +#include +#include +#include #include "secsgem/gem/data_model.hpp" @@ -300,3 +304,76 @@ TEST_CASE("spool force flag controls whether enqueue is taken") { s.set_force_spool(true); CHECK(s.force_spool()); } + +TEST_CASE("spool persistence: write, restart, replay") { + namespace fs = std::filesystem; + // Unique temp dir per test run. + auto dir = fs::temp_directory_path() / + ("spool_test_" + std::to_string(::getpid()) + "_" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count())); + fs::remove_all(dir); + + { + SpoolStore s; + s.set_spoolable_streams({5, 6}); + s.enable_persistence(dir); + CHECK(s.empty()); + s.enqueue(s2::Message(6, 11, false, s2::Item::u4(uint32_t{42}))); + s.enqueue(s2::Message(5, 1, false, s2::Item::ascii("ALARM"))); + CHECK(s.size() == 2); + // Two journal files written. + std::size_t count = 0; + for (auto& e : fs::directory_iterator(dir)) { + if (e.path().extension() == ".spool") ++count; + } + CHECK(count == 2); + } + + // Simulate restart: fresh store rehydrates from the same dir. + { + SpoolStore s; + s.set_spoolable_streams({5, 6}); + s.enable_persistence(dir); + REQUIRE(s.size() == 2); + auto drained = s.drain(); + REQUIRE(drained.size() == 2); + CHECK(drained[0].stream == 6); + CHECK(drained[1].stream == 5); + // FIFO: first-enqueued comes out first. + REQUIRE(drained[0].body.has_value()); + CHECK(std::get>(drained[0].body->storage()).front() == 42); + CHECK(drained[1].body->as_ascii() == "ALARM"); + // Files removed after drain. + std::size_t count = 0; + for (auto& e : fs::directory_iterator(dir)) { + if (e.path().extension() == ".spool") ++count; + } + CHECK(count == 0); + } + + fs::remove_all(dir); +} + +TEST_CASE("spool persistence: clear deletes files") { + namespace fs = std::filesystem; + auto dir = fs::temp_directory_path() / + ("spool_clear_" + std::to_string(::getpid()) + "_" + + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count() + 1)); + fs::remove_all(dir); + + SpoolStore s; + s.set_spoolable_streams({6}); + s.enable_persistence(dir); + s.enqueue(s2::Message(6, 11, false)); + s.enqueue(s2::Message(6, 11, false)); + CHECK(s.size() == 2); + s.clear(); + CHECK(s.empty()); + std::size_t count = 0; + for (auto& e : fs::directory_iterator(dir)) { + if (e.path().extension() == ".spool") ++count; + } + CHECK(count == 0); + + fs::remove_all(dir); +}