Files
secs-gem/include/secsgem/gem/store/exceptions.hpp
T
raphael b3bde7f087 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>
2026-06-09 10:37:36 +02:00

297 lines
11 KiB
C++

#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>
#include "secsgem/gem/exception_state.hpp"
namespace secsgem::gem {
// One Exception record. EXID keys the row; EXTYPE / EXMESSAGE are E5
// §9 metadata; EXRECVRA lists the recovery-action names the host may
// pick from on S5F13. The FSM is heap-allocated so the per-exception
// state-change handler can capture a stable pointer.
struct Exception {
uint32_t exid;
std::string extype;
std::string exmessage;
std::vector<std::string> exrecvra;
std::unique_ptr<ExceptionStateMachine> fsm;
std::filesystem::path journal_path;
};
class ExceptionStore {
public:
using TransitionTableFactory =
std::function<ExceptionTransitionTable()>;
using StateChangeHandler =
std::function<void(uint32_t exid, ExceptionState from, ExceptionState to,
ExceptionEvent trigger)>;
ExceptionStore()
: factory_([] { return ExceptionTransitionTable::default_table(); }) {}
// Same `this`-capture caveat as ProcessJobStore: keep a stable address.
ExceptionStore(const ExceptionStore&) = delete;
ExceptionStore& operator=(const ExceptionStore&) = delete;
ExceptionStore(ExceptionStore&&) = delete;
ExceptionStore& operator=(ExceptionStore&&) = delete;
void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); }
void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); }
enum class PostResult {
Posted,
Denied_AlreadyExists,
};
// Equipment posts a new exception (S5F9). Returns Denied_AlreadyExists
// if an entry for `exid` is already live. Fires a synthetic NoState ->
// Posted change so the dispatcher can decide whether to actually emit
// the S5F9 (skip if alerts disabled, etc.).
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;
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);
}
return PostResult::Posted;
}
bool has(uint32_t exid) const { return by_id_.count(exid) > 0; }
const Exception* get(uint32_t exid) const {
auto it = by_id_.find(exid);
return it == by_id_.end() ? nullptr : &it->second;
}
Exception* get(uint32_t exid) {
auto it = by_id_.find(exid);
return it == by_id_.end() ? nullptr : &it->second;
}
ExceptionState state(uint32_t exid) const {
auto it = by_id_.find(exid);
return it == by_id_.end() ? ExceptionState::NoState : it->second.fsm->state();
}
// S5F13 recovery dispatch. Validates EXRECVRA matches one the post
// advertised; returns AlarmAck::Error on unknown EXID or unknown action.
AlarmAck on_recover(uint32_t exid, const std::string& exrecvra) {
auto* ex = get(exid);
if (!ex) return AlarmAck::Error;
bool known = false;
for (const auto& a : ex->exrecvra) if (a == exrecvra) { known = true; break; }
if (!known) return AlarmAck::Error;
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;
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
// entry — exceptions are not retained past clearance, matching the
// E5 §9.3 model where Cleared is the lifecycle's terminus.
bool fire_internal(uint32_t exid, ExceptionEvent event) {
auto* ex = get(exid);
if (!ex) return false;
const bool fired = ex->fsm->on_internal(event);
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;
}
std::size_t size() const { return by_id_.size(); }
std::vector<uint32_t> ids() const {
std::vector<uint32_t> out;
out.reserve(by_id_.size());
for (const auto& kv : by_id_) out.push_back(kv.first);
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