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

persistence: CarrierStore + LoadPortStore enable_persistence(dir)

Browse Source 
Mirrors SpoolStore: per-record file with atomic .tmp+rename, magic+
version-prefixed binary layout, replay on enable, delete on remove.
FSMs gain a restore_state() that bypasses the transition table and
handlers since a replay isn't a transition.

Six new tests cover write+restart+replay across every CIDS/CSMS/CAS
axis, remove-deletes-journal, malformed-record drop-not-poison, and
the persistence-disabled no-op path.

Closes #1 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:25:50 +02:00
parent 29f646c7ca
commit f56639ba17
5 changed files with 571 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/secsgem/gem/store/carriers.hpp
+342 -43
View File
@@ -1,11 +1,16 @@
#pragma once
#include <algorithm>
#include <cstdint>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <system_error>
#include <utility>
#include <vector>
@@ -27,12 +32,14 @@ struct Carrier {
uint8_t port_id = 0; // 0 = unbound; ports are 1-based
std::vector<CarrierSlot> slots; // length == capacity (typ. 25)
std::unique_ptr<CarrierStateMachine> fsm;
std::filesystem::path journal_path; // empty when persistence disabled
};
struct LoadPort {
uint8_t port_id;
std::string carrierid; // empty when nothing associated
std::unique_ptr<LoadPortStateMachine> fsm;
std::filesystem::path journal_path; // empty when persistence disabled
};
// Both stores are non-movable for the same reason ProcessJobStore is —
@@ -64,23 +71,9 @@ class CarrierStore {
CreateResult create(std::string carrierid, uint8_t port_id = 0,
std::size_t capacity = 25) {
if (carriers_.count(carrierid)) return CreateResult::Denied_AlreadyExists;
auto fsm = std::make_unique<CarrierStateMachine>();
const std::string id = carrierid;
fsm->set_id_handler([this, id](CarrierIDStatus f, CarrierIDStatus t,
CarrierIDEvent e) {
if (on_id_) on_id_(id, f, t, e);
});
fsm->set_slot_map_handler([this, id](SlotMapStatus f, SlotMapStatus t,
SlotMapEvent e) {
if (on_sm_) on_sm_(id, f, t, e);
});
fsm->set_access_handler([this, id](CarrierAccessStatus f, CarrierAccessStatus t,
CarrierAccessEvent e) {
if (on_acc_) on_acc_(id, f, t, e);
});
carriers_.emplace(carrierid, Carrier{carrierid, port_id,
std::vector<CarrierSlot>(capacity),
std::move(fsm)});
install_(carrierid, port_id, std::vector<CarrierSlot>(capacity),
/*restore_id=*/std::nullopt, std::nullopt, std::nullopt);
if (persistent_) write_record_(carrierid);
return CreateResult::Created;
}
@@ -96,18 +89,43 @@ class CarrierStore {
bool fire_id_event(const std::string& cid, CarrierIDEvent e) {
auto* c = get(cid);
return c && c->fsm->on_id_event(e);
if (!(c && c->fsm->on_id_event(e))) return false;
if (persistent_) write_record_(cid);
return true;
}
bool fire_slot_map_event(const std::string& cid, SlotMapEvent e) {
auto* c = get(cid);
return c && c->fsm->on_slot_map_event(e);
if (!(c && c->fsm->on_slot_map_event(e))) return false;
if (persistent_) write_record_(cid);
return true;
}
bool fire_access_event(const std::string& cid, CarrierAccessEvent e) {
auto* c = get(cid);
return c && c->fsm->on_access_event(e);
if (!(c && c->fsm->on_access_event(e))) return false;
if (persistent_) write_record_(cid);
return true;
}
bool remove(const std::string& cid) { return carriers_.erase(cid) > 0; }
// Application sets slots[i].state directly; persistence sync is opt-in
// through this helper so the journal records the new slot map.
void set_slot_state(const std::string& cid, std::size_t slot_idx,
uint8_t state) {
auto* c = get(cid);
if (!c || slot_idx >= c->slots.size()) return;
c->slots[slot_idx].state = state;
if (persistent_) write_record_(cid);
}
bool remove(const std::string& cid) {
auto it = carriers_.find(cid);
if (it == carriers_.end()) return false;
if (persistent_ && !it->second.journal_path.empty()) {
std::error_code ec;
std::filesystem::remove(it->second.journal_path, ec);
}
carriers_.erase(it);
return true;
}
std::size_t size() const { return carriers_.size(); }
std::vector<std::string> ids() const {
std::vector<std::string> out;
@@ -116,11 +134,174 @@ class CarrierStore {
return out;
}
// ---- Persistence ----------------------------------------------------
// Mirrors SpoolStore::enable_persistence: scan `dir` for *.car records,
// replay them into the in-memory map, then keep the directory in sync
// on subsequent create / fire_* / remove calls. Atomic writes via
// .tmp + rename. Records that fail magic/version are silently dropped
// so a single bad file can't poison the whole store.
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() != ".car") 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->carrierid, rec->port_id, std::move(rec->slots),
rec->id_state, rec->sm_state, rec->acc_state);
auto* c = get(rec->carrierid);
if (c) c->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 carrierid;
uint8_t port_id;
std::vector<CarrierSlot> slots;
CarrierIDStatus id_state;
SlotMapStatus sm_state;
CarrierAccessStatus acc_state;
};
// Carrier record layout, big-endian throughout:
// [u8 magic = 0xC4][u8 version = 1]
// [u8 id_state][u8 sm_state][u8 acc_state][u8 port_id]
// [u16 slot_count][slot_count × u8 slot.state]
// [u16 cid_len][cid_len × u8 carrierid]
static constexpr uint8_t kMagic = 0xC4;
static constexpr uint8_t kVersion = 0x01;
void install_(const std::string& carrierid, uint8_t port_id,
std::vector<CarrierSlot> slots,
std::optional<CarrierIDStatus> id_state,
std::optional<SlotMapStatus> sm_state,
std::optional<CarrierAccessStatus> acc_state) {
auto fsm = std::make_unique<CarrierStateMachine>();
if (id_state || sm_state || acc_state) {
fsm->restore_state(id_state.value_or(CarrierIDStatus::NotConfirmed),
sm_state.value_or(SlotMapStatus::NotRead),
acc_state.value_or(CarrierAccessStatus::NotAccessed));
}
const std::string id = carrierid;
fsm->set_id_handler([this, id](CarrierIDStatus f, CarrierIDStatus t,
CarrierIDEvent e) {
if (on_id_) on_id_(id, f, t, e);
});
fsm->set_slot_map_handler([this, id](SlotMapStatus f, SlotMapStatus t,
SlotMapEvent e) {
if (on_sm_) on_sm_(id, f, t, e);
});
fsm->set_access_handler([this, id](CarrierAccessStatus f, CarrierAccessStatus t,
CarrierAccessEvent e) {
if (on_acc_) on_acc_(id, f, t, e);
});
carriers_.emplace(carrierid, Carrier{carrierid, port_id, std::move(slots),
std::move(fsm), {}});
}
void write_record_(const std::string& carrierid) {
namespace fs = std::filesystem;
auto* c = get(carrierid);
if (!c) return;
fs::path target = c->journal_path;
if (target.empty()) {
char namebuf[32];
std::snprintf(namebuf, sizeof(namebuf), "%010llu.car",
static_cast<unsigned long long>(next_seq_++));
target = journal_dir_ / namebuf;
c->journal_path = target;
}
fs::path tmp = target;
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>(c->fsm->id_status()),
static_cast<uint8_t>(c->fsm->slot_map_status()),
static_cast<uint8_t>(c->fsm->access_status()),
c->port_id};
out.write(reinterpret_cast<const char*>(header), sizeof(header));
auto write_be16 = [&](uint16_t v) {
uint8_t b[2] = {static_cast<uint8_t>(v >> 8), static_cast<uint8_t>(v)};
out.write(reinterpret_cast<const char*>(b), 2);
};
write_be16(static_cast<uint16_t>(c->slots.size()));
for (const auto& s : c->slots) {
out.put(static_cast<char>(s.state));
}
write_be16(static_cast<uint16_t>(c->carrierid.size()));
out.write(c->carrierid.data(),
static_cast<std::streamsize>(c->carrierid.size()));
out.flush();
out.close();
std::error_code ec;
fs::rename(tmp, target, 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.id_state = static_cast<CarrierIDStatus>(header[2]);
r.sm_state = static_cast<SlotMapStatus>(header[3]);
r.acc_state = static_cast<CarrierAccessStatus>(header[4]);
r.port_id = header[5];
auto read_be16 = [&](uint16_t& v) {
uint8_t b[2];
in.read(reinterpret_cast<char*>(b), 2);
if (!in) return false;
v = static_cast<uint16_t>((uint16_t(b[0]) << 8) | b[1]);
return true;
};
uint16_t slot_count = 0;
if (!read_be16(slot_count)) return std::nullopt;
r.slots.resize(slot_count);
for (auto& s : r.slots) {
int c = in.get();
if (c == std::char_traits<char>::eof()) return std::nullopt;
s.state = static_cast<uint8_t>(c);
}
uint16_t cid_len = 0;
if (!read_be16(cid_len)) return std::nullopt;
r.carrierid.resize(cid_len);
in.read(r.carrierid.data(), cid_len);
if (!in) return std::nullopt;
return r;
}
std::map<std::string, Carrier> carriers_;
IDChangeHandler on_id_;
SlotMapChangeHandler on_sm_;
AccessChangeHandler on_acc_;
bool persistent_ = false;
std::filesystem::path journal_dir_;
uint64_t next_seq_ = 0;
};
class LoadPortStore {
@@ -148,24 +329,8 @@ class LoadPortStore {
// Idempotent: re-create a port_id is a no-op.
bool create(uint8_t port_id) {
if (ports_.count(port_id)) return false;
auto fsm = std::make_unique<LoadPortStateMachine>();
const uint8_t pid = port_id;
fsm->set_transfer_handler([this, pid](LoadPortTransferState f,
LoadPortTransferState t,
LoadPortTransferEvent e) {
if (on_tx_) on_tx_(pid, f, t, e);
});
fsm->set_reservation_handler([this, pid](LoadPortReservationStatus f,
LoadPortReservationStatus t,
LoadPortReservationEvent e) {
if (on_rs_) on_rs_(pid, f, t, e);
});
fsm->set_association_handler([this, pid](LoadPortAssociationStatus f,
LoadPortAssociationStatus t,
LoadPortAssociationEvent e) {
if (on_as_) on_as_(pid, f, t, e);
});
ports_.emplace(port_id, LoadPort{port_id, "", std::move(fsm)});
install_(port_id, "", std::nullopt, std::nullopt, std::nullopt);
if (persistent_) write_record_(port_id);
return true;
}
@@ -183,23 +348,31 @@ class LoadPortStore {
auto* p = get(pid);
if (!p) return false;
p->carrierid = std::move(carrierid);
return p->fsm->on_association_event(LoadPortAssociationEvent::Associate);
bool ok = p->fsm->on_association_event(LoadPortAssociationEvent::Associate);
if (persistent_) write_record_(pid);
return ok;
}
bool disassociate(uint8_t pid) {
auto* p = get(pid);
if (!p) return false;
p->carrierid.clear();
return p->fsm->on_association_event(LoadPortAssociationEvent::Disassociate);
bool ok = p->fsm->on_association_event(LoadPortAssociationEvent::Disassociate);
if (persistent_) write_record_(pid);
return ok;
}
bool fire_transfer_event(uint8_t pid, LoadPortTransferEvent e) {
auto* p = get(pid);
return p && p->fsm->on_transfer_event(e);
if (!(p && p->fsm->on_transfer_event(e))) return false;
if (persistent_) write_record_(pid);
return true;
}
bool fire_reservation_event(uint8_t pid, LoadPortReservationEvent e) {
auto* p = get(pid);
return p && p->fsm->on_reservation_event(e);
if (!(p && p->fsm->on_reservation_event(e))) return false;
if (persistent_) write_record_(pid);
return true;
}
std::size_t size() const { return ports_.size(); }
@@ -210,11 +383,137 @@ class LoadPortStore {
return out;
}
// ---- Persistence ----------------------------------------------------
// One file per port: `<port_id 0-padded>.lp`. Port IDs are 1–255 so
// we can use the port id directly as the filename stem.
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);
for (auto& entry : fs::directory_iterator(journal_dir_, ec)) {
if (!entry.is_regular_file()) continue;
const auto& p = entry.path();
if (p.extension() != ".lp") continue;
auto rec = load_record_(p);
if (!rec) { fs::remove(p, ec); continue; }
install_(rec->port_id, rec->carrierid, rec->tx_state, rec->rs_state,
rec->as_state);
auto* lp = get(rec->port_id);
if (lp) lp->journal_path = p;
}
}
bool persistence_enabled() const { return persistent_; }
std::filesystem::path journal_dir() const { return journal_dir_; }
private:
struct Record {
uint8_t port_id;
std::string carrierid;
LoadPortTransferState tx_state;
LoadPortReservationStatus rs_state;
LoadPortAssociationStatus as_state;
};
// Load-port record:
// [u8 magic = 0xC5][u8 version = 1]
// [u8 port_id][u8 tx_state][u8 rs_state][u8 as_state]
// [u16 cid_len][cid_len × u8 carrierid]
static constexpr uint8_t kMagic = 0xC5;
static constexpr uint8_t kVersion = 0x01;
void install_(uint8_t port_id, std::string carrierid,
std::optional<LoadPortTransferState> tx,
std::optional<LoadPortReservationStatus> rs,
std::optional<LoadPortAssociationStatus> as) {
auto fsm = std::make_unique<LoadPortStateMachine>();
if (tx || rs || as) {
fsm->restore_state(tx.value_or(LoadPortTransferState::InService),
rs.value_or(LoadPortReservationStatus::NotReserved),
as.value_or(LoadPortAssociationStatus::NotAssociated));
}
const uint8_t pid = port_id;
fsm->set_transfer_handler([this, pid](LoadPortTransferState f,
LoadPortTransferState t,
LoadPortTransferEvent e) {
if (on_tx_) on_tx_(pid, f, t, e);
});
fsm->set_reservation_handler([this, pid](LoadPortReservationStatus f,
LoadPortReservationStatus t,
LoadPortReservationEvent e) {
if (on_rs_) on_rs_(pid, f, t, e);
});
fsm->set_association_handler([this, pid](LoadPortAssociationStatus f,
LoadPortAssociationStatus t,
LoadPortAssociationEvent e) {
if (on_as_) on_as_(pid, f, t, e);
});
ports_.emplace(port_id, LoadPort{port_id, std::move(carrierid),
std::move(fsm), {}});
}
void write_record_(uint8_t pid) {
namespace fs = std::filesystem;
auto* lp = get(pid);
if (!lp) return;
if (lp->journal_path.empty()) {
char namebuf[16];
std::snprintf(namebuf, sizeof(namebuf), "%03u.lp", unsigned(pid));
lp->journal_path = journal_dir_ / namebuf;
}
fs::path tmp = lp->journal_path;
tmp += ".tmp";
std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
if (!out) return;
uint8_t header[6] = {
kMagic, kVersion, pid,
static_cast<uint8_t>(lp->fsm->transfer_state()),
static_cast<uint8_t>(lp->fsm->reservation_state()),
static_cast<uint8_t>(lp->fsm->association_state())};
out.write(reinterpret_cast<const char*>(header), sizeof(header));
uint16_t cid_len = static_cast<uint16_t>(lp->carrierid.size());
uint8_t lb[2] = {static_cast<uint8_t>(cid_len >> 8),
static_cast<uint8_t>(cid_len)};
out.write(reinterpret_cast<const char*>(lb), 2);
out.write(lp->carrierid.data(),
static_cast<std::streamsize>(lp->carrierid.size()));
out.flush();
out.close();
std::error_code ec;
fs::rename(tmp, lp->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.port_id = header[2];
r.tx_state = static_cast<LoadPortTransferState>(header[3]);
r.rs_state = static_cast<LoadPortReservationStatus>(header[4]);
r.as_state = static_cast<LoadPortAssociationStatus>(header[5]);
uint8_t lb[2];
in.read(reinterpret_cast<char*>(lb), 2);
if (!in) return std::nullopt;
uint16_t cid_len = static_cast<uint16_t>((uint16_t(lb[0]) << 8) | lb[1]);
r.carrierid.resize(cid_len);
in.read(r.carrierid.data(), cid_len);
if (cid_len && !in) return std::nullopt;
return r;
}
std::map<uint8_t, LoadPort> ports_;
TransferChangeHandler on_tx_;
ReservationChangeHandler on_rs_;
AssociationChangeHandler on_as_;
bool persistent_ = false;
std::filesystem::path journal_dir_;
};
} // namespace secsgem::gem