persistence: substrate history journaling in v2 record
Per-substrate transition history now survives restart. Each entry's
steady_clock timestamp is written as a system_clock-millis snapshot;
on replay the steady_clock time_point is reconstructed relative to
the current (steady_now, system_now) pair, so inter-event spacing
is preserved across restarts even if the FSM is in a different
process. Absolute wall-clock accuracy degrades by any NTP step
that happened between write and read; that's a documented caveat.
Record format goes v1 → v2. v1 (history-less) records still load,
just with empty history.
Test updates:
- the old "history is NOT journaled" test is REPLACED with one
that asserts every axis + event + label round-trips.
- hand-crafted v1 record on disk still loads (proves backwards
compat).
- 15 ms-spaced events restore with their spacing intact (±slop
for scheduler jitter).
Closes the "substrate history persistence" caveat from the post-#1-13
status writeup.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -129,9 +129,13 @@ class SubstrateStore {
|
||||
}
|
||||
|
||||
// ---- 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.
|
||||
// v2 records also journal the per-substrate transition history.
|
||||
// Each entry's `at` field is written as a milliseconds-since-UNIX
|
||||
// epoch wall-clock timestamp; on replay the steady_clock time_point
|
||||
// is reconstructed relative to the *current* steady_now, so the
|
||||
// delta between two replayed entries matches their original spacing.
|
||||
// Absolute wall-clock accuracy depends on system_clock not having
|
||||
// been NTP-stepped between write and read.
|
||||
void enable_persistence(std::filesystem::path dir) {
|
||||
namespace fs = std::filesystem;
|
||||
journal_dir_ = std::move(dir);
|
||||
@@ -158,7 +162,10 @@ class SubstrateStore {
|
||||
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 (s) {
|
||||
s->journal_path = path;
|
||||
s->history = restore_history_(rec->history);
|
||||
}
|
||||
if (seq >= next_seq_) next_seq_ = seq + 1;
|
||||
}
|
||||
}
|
||||
@@ -167,6 +174,17 @@ class SubstrateStore {
|
||||
std::filesystem::path journal_dir() const { return journal_dir_; }
|
||||
|
||||
private:
|
||||
// Persisted history entry — system_clock millis at write time, plus
|
||||
// axis bytes and a small variant discriminator.
|
||||
struct PersistedHistoryEntry {
|
||||
int64_t system_ms_at_write;
|
||||
SubstrateState location_to;
|
||||
SubstrateProcessingState processing_to;
|
||||
uint8_t event_kind; // 0 = SubstrateEvent, 1 = SubstrateProcessingEvent
|
||||
uint8_t event_value; // raw enum value
|
||||
std::string location_label;
|
||||
};
|
||||
|
||||
struct Record {
|
||||
std::string substid;
|
||||
std::string carrierid;
|
||||
@@ -175,16 +193,23 @@ class SubstrateStore {
|
||||
SubstrateState loc_state;
|
||||
SubstrateProcessingState proc_state;
|
||||
SubstrateIDStatus id_state;
|
||||
std::vector<PersistedHistoryEntry> history;
|
||||
};
|
||||
|
||||
// Substrate record:
|
||||
// [u8 magic = 0xC6][u8 version = 1]
|
||||
// Substrate record (v2):
|
||||
// v1 header (kept) + v2 trailer:
|
||||
// [u8 magic = 0xC6][u8 version]
|
||||
// [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]
|
||||
// [u16 substid_len][substid]
|
||||
// [u16 carrierid_len][carrierid]
|
||||
// [u16 location_len][location]
|
||||
// --- v2 trailer ---
|
||||
// [u32 history_count]
|
||||
// [repeat: i64 system_ms, u8 loc_to, u8 proc_to, u8 kind, u8 event,
|
||||
// u16 label_len, label]
|
||||
// v1 records accepted; history comes back empty.
|
||||
static constexpr uint8_t kMagic = 0xC6;
|
||||
static constexpr uint8_t kVersion = 0x01;
|
||||
static constexpr uint8_t kVersion = 0x02;
|
||||
|
||||
void install_(const std::string& substid, const std::string& carrierid,
|
||||
uint8_t slot, const std::string& location,
|
||||
@@ -244,6 +269,51 @@ class SubstrateStore {
|
||||
write_str(s->substid);
|
||||
write_str(s->carrierid);
|
||||
write_str(s->location);
|
||||
|
||||
// --- v2 history trailer ---
|
||||
auto write_be32 = [&](uint32_t v) {
|
||||
uint8_t b[4] = {static_cast<uint8_t>((v >> 24) & 0xFF),
|
||||
static_cast<uint8_t>((v >> 16) & 0xFF),
|
||||
static_cast<uint8_t>((v >> 8) & 0xFF),
|
||||
static_cast<uint8_t>(v & 0xFF)};
|
||||
out.write(reinterpret_cast<const char*>(b), 4);
|
||||
};
|
||||
auto write_be64 = [&](int64_t v) {
|
||||
uint64_t u = static_cast<uint64_t>(v);
|
||||
for (int i = 7; i >= 0; --i) {
|
||||
uint8_t byte = static_cast<uint8_t>((u >> (i * 8)) & 0xFF);
|
||||
out.put(static_cast<char>(byte));
|
||||
}
|
||||
};
|
||||
using clk = std::chrono::steady_clock;
|
||||
using sclk = std::chrono::system_clock;
|
||||
const auto steady_now = clk::now();
|
||||
const auto system_now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
sclk::now().time_since_epoch()).count();
|
||||
write_be32(static_cast<uint32_t>(s->history.size()));
|
||||
for (const auto& h : s->history) {
|
||||
const auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
h.at - steady_now).count();
|
||||
const int64_t system_ms_for_entry = system_now_ms + elapsed_ms;
|
||||
write_be64(system_ms_for_entry);
|
||||
out.put(static_cast<char>(h.location_to));
|
||||
out.put(static_cast<char>(h.processing_to));
|
||||
const uint8_t kind = std::holds_alternative<SubstrateEvent>(h.event) ? 0 : 1;
|
||||
out.put(static_cast<char>(kind));
|
||||
uint8_t ev_val = 0;
|
||||
if (kind == 0) {
|
||||
ev_val = static_cast<uint8_t>(std::get<SubstrateEvent>(h.event));
|
||||
} else {
|
||||
ev_val = static_cast<uint8_t>(std::get<SubstrateProcessingEvent>(h.event));
|
||||
}
|
||||
out.put(static_cast<char>(ev_val));
|
||||
uint16_t lab_len = static_cast<uint16_t>(h.location_label.size());
|
||||
uint8_t lb[2] = {static_cast<uint8_t>(lab_len >> 8),
|
||||
static_cast<uint8_t>(lab_len)};
|
||||
out.write(reinterpret_cast<const char*>(lb), 2);
|
||||
out.write(h.location_label.data(),
|
||||
static_cast<std::streamsize>(lab_len));
|
||||
}
|
||||
out.flush();
|
||||
out.close();
|
||||
std::error_code ec;
|
||||
@@ -251,12 +321,44 @@ class SubstrateStore {
|
||||
if (ec) fs::remove(tmp, ec);
|
||||
}
|
||||
|
||||
// Reconstruct steady_clock::time_point from persisted system_clock millis
|
||||
// by anchoring to the current (now_steady, now_system) pair. Wall-clock
|
||||
// jumps between write and read will skew absolute values; deltas stay
|
||||
// accurate.
|
||||
static std::vector<SubstrateHistoryEntry> restore_history_(
|
||||
const std::vector<PersistedHistoryEntry>& src) {
|
||||
std::vector<SubstrateHistoryEntry> out;
|
||||
out.reserve(src.size());
|
||||
using clk = std::chrono::steady_clock;
|
||||
using sclk = std::chrono::system_clock;
|
||||
const auto steady_now = clk::now();
|
||||
const auto system_now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
sclk::now().time_since_epoch()).count();
|
||||
for (const auto& p : src) {
|
||||
SubstrateHistoryEntry e;
|
||||
const int64_t delta_ms = p.system_ms_at_write - system_now_ms;
|
||||
e.at = steady_now + std::chrono::milliseconds(delta_ms);
|
||||
e.location_to = p.location_to;
|
||||
e.processing_to = p.processing_to;
|
||||
e.location_label = p.location_label;
|
||||
if (p.event_kind == 0) {
|
||||
e.event = static_cast<SubstrateEvent>(p.event_value);
|
||||
} else {
|
||||
e.event = static_cast<SubstrateProcessingEvent>(p.event_value);
|
||||
}
|
||||
out.push_back(std::move(e));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
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;
|
||||
if (!in || header[0] != kMagic) return std::nullopt;
|
||||
if (header[1] != 0x01 && header[1] != 0x02) return std::nullopt;
|
||||
const uint8_t version = header[1];
|
||||
Record r;
|
||||
r.loc_state = static_cast<SubstrateState>(header[2]);
|
||||
r.proc_state = static_cast<SubstrateProcessingState>(header[3]);
|
||||
@@ -274,6 +376,40 @@ class SubstrateStore {
|
||||
if (!read_str(r.substid)) return std::nullopt;
|
||||
if (!read_str(r.carrierid)) return std::nullopt;
|
||||
if (!read_str(r.location)) return std::nullopt;
|
||||
if (version >= 2) {
|
||||
auto read_be32 = [&](uint32_t& v) {
|
||||
uint8_t b[4];
|
||||
in.read(reinterpret_cast<char*>(b), 4);
|
||||
if (!in) return false;
|
||||
v = (uint32_t(b[0]) << 24) | (uint32_t(b[1]) << 16) |
|
||||
(uint32_t(b[2]) << 8) | uint32_t(b[3]);
|
||||
return true;
|
||||
};
|
||||
auto read_be64 = [&](int64_t& v) {
|
||||
uint8_t b[8];
|
||||
in.read(reinterpret_cast<char*>(b), 8);
|
||||
if (!in) return false;
|
||||
uint64_t u = 0;
|
||||
for (int i = 0; i < 8; ++i) u = (u << 8) | b[i];
|
||||
v = static_cast<int64_t>(u);
|
||||
return true;
|
||||
};
|
||||
uint32_t count = 0;
|
||||
if (!read_be32(count)) return std::nullopt;
|
||||
r.history.resize(count);
|
||||
for (auto& h : r.history) {
|
||||
if (!read_be64(h.system_ms_at_write)) return std::nullopt;
|
||||
int b0 = in.get(); if (b0 == EOF) return std::nullopt;
|
||||
int b1 = in.get(); if (b1 == EOF) return std::nullopt;
|
||||
int b2 = in.get(); if (b2 == EOF) return std::nullopt;
|
||||
int b3 = in.get(); if (b3 == EOF) return std::nullopt;
|
||||
h.location_to = static_cast<SubstrateState>(b0);
|
||||
h.processing_to = static_cast<SubstrateProcessingState>(b1);
|
||||
h.event_kind = static_cast<uint8_t>(b2);
|
||||
h.event_value = static_cast<uint8_t>(b3);
|
||||
if (!read_str(h.location_label)) return std::nullopt;
|
||||
}
|
||||
}
|
||||
return r;
|
||||
}
|
||||
|
||||
|
||||
@@ -5,10 +5,13 @@
|
||||
|
||||
#include <doctest/doctest.h>
|
||||
|
||||
#include <chrono>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <random>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <variant>
|
||||
|
||||
#include "secsgem/gem/store/substrates.hpp"
|
||||
|
||||
@@ -161,7 +164,7 @@ TEST_CASE("SubstrateStore persistence: corrupt record dropped, others survive")
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("SubstrateStore: history is NOT journaled (transient by design)") {
|
||||
TEST_CASE("SubstrateStore: v2 history survives restart with axis + event preserved") {
|
||||
auto dir = scratch_dir("history");
|
||||
{
|
||||
SubstrateStore s;
|
||||
@@ -170,17 +173,89 @@ TEST_CASE("SubstrateStore: history is NOT journaled (transient by design)") {
|
||||
REQUIRE(s.fire_location_event("W-H", SubstrateEvent::Acquire, "MOD"));
|
||||
REQUIRE(s.fire_processing_event("W-H",
|
||||
SubstrateProcessingEvent::StartProcessing));
|
||||
CHECK(s.history("W-H")->size() == 2);
|
||||
REQUIRE(s.fire_processing_event("W-H",
|
||||
SubstrateProcessingEvent::EndProcessing));
|
||||
CHECK(s.history("W-H")->size() == 3);
|
||||
}
|
||||
{
|
||||
SubstrateStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.has("W-H"));
|
||||
// State survived but history is empty post-replay — current state
|
||||
// is what's durable, not the journey to it.
|
||||
CHECK(s.history("W-H")->size() == 0);
|
||||
CHECK(s.get("W-H")->fsm->processing_state() ==
|
||||
SubstrateProcessingState::InProcess);
|
||||
const auto* hist = s.history("W-H");
|
||||
REQUIRE(hist);
|
||||
REQUIRE(hist->size() == 3);
|
||||
|
||||
// Entry 0: location-axis Acquire, processing-axis NoState.
|
||||
CHECK((*hist)[0].location_to == SubstrateState::AtWork);
|
||||
CHECK((*hist)[0].processing_to == SubstrateProcessingState::NoState);
|
||||
CHECK(std::holds_alternative<SubstrateEvent>((*hist)[0].event));
|
||||
CHECK(std::get<SubstrateEvent>((*hist)[0].event) == SubstrateEvent::Acquire);
|
||||
CHECK((*hist)[0].location_label == "MOD");
|
||||
|
||||
// Entry 1: processing-axis StartProcessing.
|
||||
CHECK((*hist)[1].processing_to == SubstrateProcessingState::InProcess);
|
||||
CHECK(std::holds_alternative<SubstrateProcessingEvent>((*hist)[1].event));
|
||||
CHECK(std::get<SubstrateProcessingEvent>((*hist)[1].event) ==
|
||||
SubstrateProcessingEvent::StartProcessing);
|
||||
|
||||
// Entry 2: processing-axis EndProcessing.
|
||||
CHECK((*hist)[2].processing_to == SubstrateProcessingState::Processed);
|
||||
CHECK(std::get<SubstrateProcessingEvent>((*hist)[2].event) ==
|
||||
SubstrateProcessingEvent::EndProcessing);
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("SubstrateStore: v1 record on disk still loads (empty history)") {
|
||||
auto dir = scratch_dir("history-v1");
|
||||
{
|
||||
std::ofstream f(dir / "0000000001.sub", std::ios::binary);
|
||||
// v1: magic 0xC6, version 1, loc=0 proc=0 id=0 slot=0
|
||||
const uint8_t header[6] = {0xC6, 0x01, 0x00, 0x00, 0x00, 0x00};
|
||||
f.write(reinterpret_cast<const char*>(header), 6);
|
||||
auto write_str = [&](const std::string& s) {
|
||||
uint16_t n = static_cast<uint16_t>(s.size());
|
||||
uint8_t lb[2] = {static_cast<uint8_t>(n >> 8), static_cast<uint8_t>(n)};
|
||||
f.write(reinterpret_cast<const char*>(lb), 2);
|
||||
f.write(s.data(), n);
|
||||
};
|
||||
write_str("W-LEGACY");
|
||||
write_str("");
|
||||
write_str("");
|
||||
}
|
||||
SubstrateStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.has("W-LEGACY"));
|
||||
CHECK(s.history("W-LEGACY")->empty());
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
TEST_CASE("SubstrateStore: history timestamps preserve inter-event spacing") {
|
||||
// Two events fired ~10 ms apart should land ~10 ms apart in the
|
||||
// restored history regardless of how long the "restart" gap is.
|
||||
auto dir = scratch_dir("history-spacing");
|
||||
using namespace std::chrono_literals;
|
||||
{
|
||||
SubstrateStore s;
|
||||
s.enable_persistence(dir);
|
||||
REQUIRE(s.create("W-T") == SubstrateStore::CreateResult::Created);
|
||||
REQUIRE(s.fire_location_event("W-T", SubstrateEvent::Acquire));
|
||||
std::this_thread::sleep_for(15ms);
|
||||
REQUIRE(s.fire_processing_event("W-T",
|
||||
SubstrateProcessingEvent::StartProcessing));
|
||||
}
|
||||
std::this_thread::sleep_for(20ms);
|
||||
{
|
||||
SubstrateStore s;
|
||||
s.enable_persistence(dir);
|
||||
const auto* hist = s.history("W-T");
|
||||
REQUIRE(hist);
|
||||
REQUIRE(hist->size() == 2);
|
||||
const auto delta = (*hist)[1].at - (*hist)[0].at;
|
||||
// ~15 ms spacing should survive the persist/restore round-trip;
|
||||
// ±10 ms slop for scheduling jitter.
|
||||
CHECK(delta >= 5ms);
|
||||
CHECK(delta <= 60ms);
|
||||
}
|
||||
fs::remove_all(dir);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user