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

persistence: SubstrateStore enable_persistence(dir)

Browse Source 
Same pattern as carriers: per-substrate binary record (.sub) with
atomic .tmp+rename, replay on enable, delete on remove. Records
current state across all three E90 axes (location / processing /
ID-status), plus substid / carrierid / slot / free-form location
label. History is deliberately NOT journaled — it's an in-memory
ring buffer and rebuilding from replayed state would mislead.

Five new tests cover full-axis replay, every terminal processing
state, remove-deletes-journal, corrupt-record drop, and the
history-is-transient invariant.

Closes #2 in the test-gap backlog.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-09 10:31:54 +02:00
parent f56639ba17
commit 1548b49afd
4 changed files with 372 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/secsgem/gem/store/substrates.hpp
+176 -19
View File
@@ -2,10 +2,15 @@
#include <chrono>
#include <cstdint>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <system_error>
#include <utility>
#include <variant>
#include <vector>
@@ -34,6 +39,7 @@ struct Substrate {
std::string location; // free-form: equipment-defined module name
std::vector<SubstrateHistoryEntry> history;
std::unique_ptr<SubstrateStateMachine> fsm;
std::filesystem::path journal_path; // empty when persistence disabled
};
class SubstrateStore {
@@ -67,22 +73,9 @@ class SubstrateStore {
CreateResult create(std::string substid, std::string carrierid = "",
uint8_t slot = 0) {
if (subs_.count(substid)) return CreateResult::Denied_AlreadyExists;
auto fsm = std::make_unique<SubstrateStateMachine>();
const std::string id = substid;
fsm->set_location_handler(
[this, id](SubstrateState f, SubstrateState t, SubstrateEvent e) {
record_history(id, t, SubstrateProcessingState::NoState, e);
if (on_loc_) on_loc_(id, f, t, e);
});
fsm->set_processing_handler(
[this, id](SubstrateProcessingState f, SubstrateProcessingState t,
SubstrateProcessingEvent e) {
record_history(id, SubstrateState::NoState, t, e);
if (on_proc_) on_proc_(id, f, t, e);
});
subs_.emplace(substid,
Substrate{substid, std::move(carrierid), slot, "",
{}, std::move(fsm)});
install_(substid, carrierid, slot, "", std::nullopt, std::nullopt,
std::nullopt);
if (persistent_) write_record_(substid);
return CreateResult::Created;
}
@@ -101,14 +94,27 @@ class SubstrateStore {
auto* s = get(id);
if (!s) return false;
if (!new_location.empty()) s->location = std::move(new_location);
return s->fsm->on_location_event(e);
bool ok = s->fsm->on_location_event(e);
if (persistent_) write_record_(id);
return ok;
}
bool fire_processing_event(const std::string& id, SubstrateProcessingEvent e) {
auto* s = get(id);
return s && s->fsm->on_processing_event(e);
if (!(s && s->fsm->on_processing_event(e))) return false;
if (persistent_) write_record_(id);
return true;
}
bool remove(const std::string& id) { return subs_.erase(id) > 0; }
bool remove(const std::string& id) {
auto it = subs_.find(id);
if (it == subs_.end()) return false;
if (persistent_ && !it->second.journal_path.empty()) {
std::error_code ec;
std::filesystem::remove(it->second.journal_path, ec);
}
subs_.erase(it);
return true;
}
std::size_t size() const { return subs_.size(); }
std::vector<std::string> ids() const {
std::vector<std::string> out;
@@ -122,7 +128,155 @@ class SubstrateStore {
return s ? &s->history : nullptr;
}
// ---- Persistence ----------------------------------------------------
// Records *current state* per substrate. History is not journaled —
// it's a transient in-memory ring buffer and rebuilding it from
// replayed state would be misleading.
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() != ".sub") 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->substid, rec->carrierid, rec->slot, rec->location,
rec->loc_state, rec->proc_state, rec->id_state);
auto* s = get(rec->substid);
if (s) s->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 {
std::string substid;
std::string carrierid;
uint8_t slot;
std::string location;
SubstrateState loc_state;
SubstrateProcessingState proc_state;
SubstrateIDStatus id_state;
};
// Substrate record:
// [u8 magic = 0xC6][u8 version = 1]
// [u8 loc_state][u8 proc_state][u8 id_state][u8 slot]
// [u16 substid_len][substid_len × u8]
// [u16 carrierid_len][carrierid_len × u8]
// [u16 location_len][location_len × u8]
static constexpr uint8_t kMagic = 0xC6;
static constexpr uint8_t kVersion = 0x01;
void install_(const std::string& substid, const std::string& carrierid,
uint8_t slot, const std::string& location,
std::optional<SubstrateState> loc,
std::optional<SubstrateProcessingState> proc,
std::optional<SubstrateIDStatus> id) {
auto fsm = std::make_unique<SubstrateStateMachine>();
if (loc || proc || id) {
fsm->restore_state(loc.value_or(SubstrateState::AtSource),
proc.value_or(SubstrateProcessingState::NeedsProcessing),
id.value_or(SubstrateIDStatus::NotConfirmed));
}
const std::string keyed = substid;
fsm->set_location_handler(
[this, keyed](SubstrateState f, SubstrateState t, SubstrateEvent e) {
record_history(keyed, t, SubstrateProcessingState::NoState, e);
if (on_loc_) on_loc_(keyed, f, t, e);
});
fsm->set_processing_handler(
[this, keyed](SubstrateProcessingState f, SubstrateProcessingState t,
SubstrateProcessingEvent e) {
record_history(keyed, SubstrateState::NoState, t, e);
if (on_proc_) on_proc_(keyed, f, t, e);
});
subs_.emplace(substid,
Substrate{substid, carrierid, slot, location, {},
std::move(fsm), {}});
}
void write_record_(const std::string& substid) {
namespace fs = std::filesystem;
auto* s = get(substid);
if (!s) return;
if (s->journal_path.empty()) {
char namebuf[32];
std::snprintf(namebuf, sizeof(namebuf), "%010llu.sub",
static_cast<unsigned long long>(next_seq_++));
s->journal_path = journal_dir_ / namebuf;
}
fs::path tmp = s->journal_path;
tmp += ".tmp";
std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
if (!out) return;
uint8_t header[6] = {
kMagic, kVersion,
static_cast<uint8_t>(s->fsm->location_state()),
static_cast<uint8_t>(s->fsm->processing_state()),
static_cast<uint8_t>(s->fsm->id_state()),
s->slot};
out.write(reinterpret_cast<const char*>(header), sizeof(header));
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(s->substid);
write_str(s->carrierid);
write_str(s->location);
out.flush();
out.close();
std::error_code ec;
fs::rename(tmp, s->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[6];
in.read(reinterpret_cast<char*>(header), sizeof(header));
if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt;
Record r;
r.loc_state = static_cast<SubstrateState>(header[2]);
r.proc_state = static_cast<SubstrateProcessingState>(header[3]);
r.id_state = static_cast<SubstrateIDStatus>(header[4]);
r.slot = header[5];
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.substid)) return std::nullopt;
if (!read_str(r.carrierid)) return std::nullopt;
if (!read_str(r.location)) return std::nullopt;
return r;
}
template <typename Event>
void record_history(const std::string& id, SubstrateState loc_to,
SubstrateProcessingState proc_to, Event ev) {
@@ -146,6 +300,9 @@ class SubstrateStore {
LocationChangeHandler on_loc_;
ProcessingChangeHandler on_proc_;
std::size_t history_limit_ = 256;
bool persistent_ = false;
std::filesystem::path journal_dir_;
uint64_t next_seq_ = 0;
};
} // namespace secsgem::gem