persistence: ExceptionStore enable_persistence(dir)
Per-EXID binary record (.ex), magic + version + atomic .tmp+rename. Records full E5 §9 lifecycle: state, EXID, EXTYPE, EXMESSAGE, and the candidate EXRECVRA list. Cleared exceptions are terminal — the FSM transitions through Cleared remove the in-memory entry AND delete the journal file (matching the existing in-memory semantics). Recovering / RecoverFailed states survive restart: the application can decide on replay whether to retry recovery or abort. Five new tests cover post+replay, Recovering-survives-restart, autonomous-clear cleanup, RecoverFailed retry post-restart, and corrupt-record drop. This completes #12 in the test-gap backlog (persistence for the four in-memory stores beyond Spool). Closes #4 in the test-gap backlog. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -110,6 +110,7 @@ add_executable(secsgem_tests
|
||||
tests/test_control_jobs.cpp
|
||||
tests/test_job_persistence.cpp
|
||||
tests/test_exceptions.cpp
|
||||
tests/test_exception_persistence.cpp
|
||||
tests/test_carrier_state.cpp
|
||||
tests/test_carriers.cpp
|
||||
tests/test_carrier_persistence.cpp
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
#pragma once
|
||||
|
||||
#include <algorithm>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <system_error>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
@@ -20,6 +27,7 @@ struct Exception {
|
||||
std::string exmessage;
|
||||
std::vector<std::string> exrecvra;
|
||||
std::unique_ptr<ExceptionStateMachine> fsm;
|
||||
std::filesystem::path journal_path;
|
||||
};
|
||||
|
||||
class ExceptionStore {
|
||||
@@ -54,14 +62,9 @@ class ExceptionStore {
|
||||
PostResult post(uint32_t exid, std::string extype, std::string exmessage,
|
||||
std::vector<std::string> exrecvra) {
|
||||
if (by_id_.count(exid)) return PostResult::Denied_AlreadyExists;
|
||||
auto fsm = std::make_unique<ExceptionStateMachine>(factory_(),
|
||||
ExceptionState::Posted);
|
||||
fsm->set_state_change_handler(
|
||||
[this, exid](ExceptionState from, ExceptionState to, ExceptionEvent trig) {
|
||||
if (on_change_) on_change_(exid, from, to, trig);
|
||||
});
|
||||
by_id_.emplace(exid, Exception{exid, std::move(extype), std::move(exmessage),
|
||||
std::move(exrecvra), std::move(fsm)});
|
||||
install_(exid, std::move(extype), std::move(exmessage),
|
||||
std::move(exrecvra), ExceptionState::Posted);
|
||||
if (persistent_) write_record_(exid);
|
||||
if (on_change_) {
|
||||
on_change_(exid, ExceptionState::NoState, ExceptionState::Posted,
|
||||
ExceptionEvent::Created);
|
||||
@@ -93,14 +96,18 @@ class ExceptionStore {
|
||||
bool known = false;
|
||||
for (const auto& a : ex->exrecvra) if (a == exrecvra) { known = true; break; }
|
||||
if (!known) return AlarmAck::Error;
|
||||
return ex->fsm->on_host_command(ExceptionEvent::Recover);
|
||||
AlarmAck ack = ex->fsm->on_host_command(ExceptionEvent::Recover);
|
||||
if (ack == AlarmAck::Accept && persistent_) write_record_(exid);
|
||||
return ack;
|
||||
}
|
||||
|
||||
// S5F17 recovery-abort dispatch.
|
||||
AlarmAck on_recover_abort(uint32_t exid) {
|
||||
auto* ex = get(exid);
|
||||
if (!ex) return AlarmAck::Error;
|
||||
return ex->fsm->on_host_command(ExceptionEvent::RecoveryAbort);
|
||||
AlarmAck ack = ex->fsm->on_host_command(ExceptionEvent::RecoveryAbort);
|
||||
if (ack == AlarmAck::Accept && persistent_) write_record_(exid);
|
||||
return ack;
|
||||
}
|
||||
|
||||
// Equipment-internal lifecycle events. Cleared transitions remove the
|
||||
@@ -110,8 +117,16 @@ class ExceptionStore {
|
||||
auto* ex = get(exid);
|
||||
if (!ex) return false;
|
||||
const bool fired = ex->fsm->on_internal(event);
|
||||
if (fired && ex->fsm->state() == ExceptionState::Cleared) {
|
||||
by_id_.erase(exid);
|
||||
if (fired) {
|
||||
if (ex->fsm->state() == ExceptionState::Cleared) {
|
||||
if (persistent_ && !ex->journal_path.empty()) {
|
||||
std::error_code ec;
|
||||
std::filesystem::remove(ex->journal_path, ec);
|
||||
}
|
||||
by_id_.erase(exid);
|
||||
} else if (persistent_) {
|
||||
write_record_(exid);
|
||||
}
|
||||
}
|
||||
return fired;
|
||||
}
|
||||
@@ -124,10 +139,158 @@ class ExceptionStore {
|
||||
return out;
|
||||
}
|
||||
|
||||
// ---- Persistence ----------------------------------------------------
|
||||
// One file per posted exception (`<seq>.ex`). Cleared exceptions are
|
||||
// removed from the store and the file is deleted; this matches the
|
||||
// lifecycle (Cleared is terminal). A crash-while-Recovering is
|
||||
// recoverable: replay leaves the exception in Recovering state and the
|
||||
// application can decide whether to retry recovery or abort.
|
||||
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);
|
||||
|
||||
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() != ".ex") 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 rec = load_record_(path);
|
||||
if (!rec) { fs::remove(path, ec); continue; }
|
||||
install_(rec->exid, std::move(rec->extype), std::move(rec->exmessage),
|
||||
std::move(rec->exrecvra), rec->state);
|
||||
auto* ex = get(rec->exid);
|
||||
if (ex) ex->journal_path = path;
|
||||
if (seq >= next_seq_) next_seq_ = seq + 1;
|
||||
}
|
||||
}
|
||||
|
||||
bool persistence_enabled() const { return persistent_; }
|
||||
std::filesystem::path journal_dir() const { return journal_dir_; }
|
||||
|
||||
private:
|
||||
struct Record {
|
||||
uint32_t exid;
|
||||
std::string extype;
|
||||
std::string exmessage;
|
||||
std::vector<std::string> exrecvra;
|
||||
ExceptionState state;
|
||||
};
|
||||
|
||||
// Exception record:
|
||||
// [u8 magic = 0xC9][u8 version = 1][u8 state]
|
||||
// [u32 exid, big-endian]
|
||||
// [u16 extype_len][extype]
|
||||
// [u16 exmessage_len][exmessage]
|
||||
// [u16 recvra_count][repeat: u16 len + bytes]
|
||||
static constexpr uint8_t kMagic = 0xC9;
|
||||
static constexpr uint8_t kVersion = 0x01;
|
||||
|
||||
void install_(uint32_t exid, std::string extype, std::string exmessage,
|
||||
std::vector<std::string> exrecvra, ExceptionState restored) {
|
||||
auto fsm = std::make_unique<ExceptionStateMachine>(factory_(), restored);
|
||||
fsm->set_state_change_handler(
|
||||
[this, exid](ExceptionState from, ExceptionState to,
|
||||
ExceptionEvent trig) {
|
||||
if (on_change_) on_change_(exid, from, to, trig);
|
||||
});
|
||||
by_id_.emplace(exid, Exception{exid, std::move(extype),
|
||||
std::move(exmessage),
|
||||
std::move(exrecvra), std::move(fsm), {}});
|
||||
}
|
||||
|
||||
void write_record_(uint32_t exid) {
|
||||
namespace fs = std::filesystem;
|
||||
auto* ex = get(exid);
|
||||
if (!ex) return;
|
||||
if (ex->journal_path.empty()) {
|
||||
char namebuf[32];
|
||||
std::snprintf(namebuf, sizeof(namebuf), "%010llu.ex",
|
||||
static_cast<unsigned long long>(next_seq_++));
|
||||
ex->journal_path = journal_dir_ / namebuf;
|
||||
}
|
||||
fs::path tmp = ex->journal_path;
|
||||
tmp += ".tmp";
|
||||
std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
|
||||
if (!out) return;
|
||||
uint8_t header[3] = {kMagic, kVersion,
|
||||
static_cast<uint8_t>(ex->fsm->state())};
|
||||
out.write(reinterpret_cast<const char*>(header), sizeof(header));
|
||||
uint8_t eb[4] = {
|
||||
static_cast<uint8_t>((exid >> 24) & 0xFF),
|
||||
static_cast<uint8_t>((exid >> 16) & 0xFF),
|
||||
static_cast<uint8_t>((exid >> 8) & 0xFF),
|
||||
static_cast<uint8_t>(exid & 0xFF)};
|
||||
out.write(reinterpret_cast<const char*>(eb), 4);
|
||||
auto write_str = [&](const std::string& str) {
|
||||
uint16_t n = static_cast<uint16_t>(str.size());
|
||||
uint8_t lb[2] = {static_cast<uint8_t>(n >> 8), static_cast<uint8_t>(n)};
|
||||
out.write(reinterpret_cast<const char*>(lb), 2);
|
||||
out.write(str.data(), static_cast<std::streamsize>(n));
|
||||
};
|
||||
write_str(ex->extype);
|
||||
write_str(ex->exmessage);
|
||||
uint16_t rc = static_cast<uint16_t>(ex->exrecvra.size());
|
||||
uint8_t rb[2] = {static_cast<uint8_t>(rc >> 8), static_cast<uint8_t>(rc)};
|
||||
out.write(reinterpret_cast<const char*>(rb), 2);
|
||||
for (const auto& a : ex->exrecvra) write_str(a);
|
||||
out.flush();
|
||||
out.close();
|
||||
std::error_code ec;
|
||||
fs::rename(tmp, ex->journal_path, ec);
|
||||
if (ec) fs::remove(tmp, ec);
|
||||
}
|
||||
|
||||
static std::optional<Record> load_record_(const std::filesystem::path& p) {
|
||||
std::ifstream in(p, std::ios::binary);
|
||||
if (!in) return std::nullopt;
|
||||
uint8_t header[3];
|
||||
in.read(reinterpret_cast<char*>(header), sizeof(header));
|
||||
if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt;
|
||||
Record r;
|
||||
r.state = static_cast<ExceptionState>(header[2]);
|
||||
uint8_t eb[4];
|
||||
in.read(reinterpret_cast<char*>(eb), 4);
|
||||
if (!in) return std::nullopt;
|
||||
r.exid = (uint32_t{eb[0]} << 24) | (uint32_t{eb[1]} << 16) |
|
||||
(uint32_t{eb[2]} << 8) | uint32_t{eb[3]};
|
||||
auto read_str = [&](std::string& out) {
|
||||
uint8_t lb[2];
|
||||
in.read(reinterpret_cast<char*>(lb), 2);
|
||||
if (!in) return false;
|
||||
uint16_t n = static_cast<uint16_t>((uint16_t(lb[0]) << 8) | lb[1]);
|
||||
out.resize(n);
|
||||
in.read(out.data(), n);
|
||||
return static_cast<bool>(in) || n == 0;
|
||||
};
|
||||
if (!read_str(r.extype)) return std::nullopt;
|
||||
if (!read_str(r.exmessage)) return std::nullopt;
|
||||
uint8_t rb[2];
|
||||
in.read(reinterpret_cast<char*>(rb), 2);
|
||||
if (!in) return std::nullopt;
|
||||
uint16_t rc = static_cast<uint16_t>((uint16_t(rb[0]) << 8) | rb[1]);
|
||||
r.exrecvra.resize(rc);
|
||||
for (auto& a : r.exrecvra) if (!read_str(a)) return std::nullopt;
|
||||
return r;
|
||||
}
|
||||
|
||||
std::map<uint32_t, Exception> by_id_;
|
||||
TransitionTableFactory factory_;
|
||||
StateChangeHandler on_change_;
|
||||
bool persistent_ = false;
|
||||
std::filesystem::path journal_dir_;
|
||||
uint64_t next_seq_ = 0;
|
||||
};
|
||||
|
||||
} // namespace secsgem::gem
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
// Persistence tests for ExceptionStore.
|
||||
|
||||
#include <doctest/doctest.h>
|
||||
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <random>
|
||||
#include <string>
|
||||
|
||||
#include "secsgem/gem/store/exceptions.hpp"
|
||||
|
||||
using namespace secsgem::gem;
|
||||
namespace fs = std::filesystem;
|
||||
|
||||
namespace {
|
||||
fs::path scratch_dir(const char* tag) {
|
||||
std::random_device rd;
|
||||
auto p = fs::temp_directory_path() /
|
||||
(std::string("secsgem-ex-") + tag + "-" + std::to_string(rd()));
|
||||
fs::remove_all(p);
|
||||
fs::create_directories(p);
|
||||
return p;
|
||||
}
|
||||
std::size_t count_with_ext(const fs::path& dir, const std::string& ext) {
|
||||
std::size_t n = 0;
|
||||
std::error_code ec;
|
||||
for (auto& e : fs::directory_iterator(dir, ec)) {
|
||||
if (e.is_regular_file() && e.path().extension() == ext) ++n;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
} // namespace
|
||||
|
||||
TEST_CASE("ExceptionStore persistence: post + replay reproduces full record") {
|
||||
auto dir = scratch_dir("post-replay");
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.post(42, "VACUUM", "lost vacuum in chamber A",
|
||||
{"PURGE", "RECOVER", "ABORT"}) ==
|
||||
ExceptionStore::PostResult::Posted);
|
||||
REQUIRE(s.post(99, "OVERTEMP", "process tube > 850C",
|
||||
{"COOL"}) ==
|
||||
ExceptionStore::PostResult::Posted);
|
||||
CHECK(count_with_ext(dir, ".ex") == 2);
|
||||
}
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.has(42));
|
||||
REQUIRE(s.has(99));
|
||||
const auto* e42 = s.get(42);
|
||||
CHECK(e42->extype == "VACUUM");
|
||||
CHECK(e42->exmessage == "lost vacuum in chamber A");
|
||||
CHECK(e42->exrecvra == std::vector<std::string>{"PURGE", "RECOVER", "ABORT"});
|
||||
CHECK(e42->fsm->state() == ExceptionState::Posted);
|
||||
|
||||
const auto* e99 = s.get(99);
|
||||
CHECK(e99->exrecvra.size() == 1);
|
||||
CHECK(e99->exrecvra[0] == "COOL");
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("ExceptionStore persistence: Recovering state survives restart") {
|
||||
auto dir = scratch_dir("recover");
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.post(1, "GAS", "leak", {"PURGE"}) ==
|
||||
ExceptionStore::PostResult::Posted);
|
||||
CHECK(s.on_recover(1, "PURGE") == AlarmAck::Accept);
|
||||
CHECK(s.state(1) == ExceptionState::Recovering);
|
||||
}
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.has(1));
|
||||
CHECK(s.state(1) == ExceptionState::Recovering);
|
||||
|
||||
// Application decides to complete recovery — state moves to Cleared,
|
||||
// which terminally removes the entry + deletes the journal file.
|
||||
REQUIRE(s.fire_internal(1, ExceptionEvent::RecoveryComplete));
|
||||
CHECK_FALSE(s.has(1));
|
||||
CHECK(count_with_ext(dir, ".ex") == 0);
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("ExceptionStore persistence: autonomous Clear deletes journal") {
|
||||
auto dir = scratch_dir("autoclear");
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.post(7, "RF", "rf fault", {"RESET"}) ==
|
||||
ExceptionStore::PostResult::Posted);
|
||||
CHECK(count_with_ext(dir, ".ex") == 1);
|
||||
REQUIRE(s.fire_internal(7, ExceptionEvent::Clear));
|
||||
CHECK_FALSE(s.has(7));
|
||||
CHECK(count_with_ext(dir, ".ex") == 0);
|
||||
}
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
CHECK(s.size() == 0);
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("ExceptionStore persistence: RecoverFailed survives restart and retry works") {
|
||||
auto dir = scratch_dir("retry");
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.post(3, "FLOW", "MFC stuck", {"RETRY", "ABORT"}) ==
|
||||
ExceptionStore::PostResult::Posted);
|
||||
CHECK(s.on_recover(3, "RETRY") == AlarmAck::Accept);
|
||||
REQUIRE(s.fire_internal(3, ExceptionEvent::RecoveryFailed));
|
||||
CHECK(s.state(3) == ExceptionState::RecoverFailed);
|
||||
}
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
CHECK(s.state(3) == ExceptionState::RecoverFailed);
|
||||
// Recovery from RecoverFailed is legal — host can retry.
|
||||
CHECK(s.on_recover(3, "RETRY") == AlarmAck::Accept);
|
||||
CHECK(s.state(3) == ExceptionState::Recovering);
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("ExceptionStore persistence: corrupt file dropped, others survive") {
|
||||
auto dir = scratch_dir("corrupt");
|
||||
{
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.post(11, "T", "M", {"R"}) == ExceptionStore::PostResult::Posted);
|
||||
}
|
||||
{
|
||||
std::ofstream f(dir / "9999999999.ex", std::ios::binary);
|
||||
const char junk[] = {0x00, 0x01, 0x02};
|
||||
f.write(junk, sizeof(junk));
|
||||
}
|
||||
ExceptionStore s;
|
||||
s.enable_persistence(dir);
|
||||
CHECK(s.has(11));
|
||||
CHECK(s.size() == 1);
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
Reference in New Issue
Block a user