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

persistence: ProcessJobStore + ControlJobStore enable_persistence(dir)

Browse Source 
Per-job binary record (.pj / .cj) with magic+version, atomic
.tmp+rename. PJ store additionally writes an order.idx index file
that preserves HOQ-aware queue position across restarts.

Rcpvars / prprocessparams (secs2::Item variants) are intentionally
out of scope for v1 — they're optional E40 trailers and need a body
codec round-trip; callers re-populate via set_e40_extras() after
restart.

Five new tests cover full lifecycle replay (Processing mid-run +
HOQ-reordered queue), dequeue-deletes-file, corrupt-record drop,
CJ state + PJ-list replay, and CJ remove cleanup.

Closes #3 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:35:26 +02:00
parent 1548b49afd
commit 1189ffc994
4 changed files with 570 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/secsgem/gem/store/control_jobs.hpp
+155 -17
View File
@@ -1,9 +1,14 @@
#pragma once
#include <cstdint>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <system_error>
#include <utility>
#include <vector>
@@ -16,6 +21,7 @@ struct ControlJob {
std::string ctljobid;
std::vector<std::string> prjobids;
std::unique_ptr<ControlJobStateMachine> fsm;
std::filesystem::path journal_path;
};
class ControlJobStore {
@@ -49,21 +55,10 @@ class ControlJobStore {
for (const auto& id : prjobids) {
if (!pj_exists(id)) return CreateResult::Denied_UnknownPRJob;
}
auto fsm = std::make_unique<ControlJobStateMachine>(factory_(),
ControlJobState::Queued);
const std::string id_for_handler = ctljobid;
install_(ctljobid, std::move(prjobids), ControlJobState::Queued);
if (persistent_) write_record_(ctljobid);
if (on_change_) {
auto cb = on_change_;
fsm->set_state_change_handler(
[cb, id_for_handler](ControlJobState from, ControlJobState to,
ControlJobEvent trig) {
cb(id_for_handler, from, to, trig);
});
}
jobs_.emplace(ctljobid, ControlJob{ctljobid, std::move(prjobids),
std::move(fsm)});
if (on_change_) {
on_change_(id_for_handler, ControlJobState::NoState,
on_change_(ctljobid, ControlJobState::NoState,
ControlJobState::Queued, ControlJobEvent::Created);
}
return CreateResult::Created;
@@ -90,17 +85,28 @@ class ControlJobStore {
HostCmdAck on_host_command(const std::string& ctljobid, ControlJobEvent event) {
auto* cj = get(ctljobid);
if (!cj) return HostCmdAck::InvalidObject;
return cj->fsm->on_host_command(event);
HostCmdAck ack = cj->fsm->on_host_command(event);
if (ack == HostCmdAck::Accept && persistent_) write_record_(ctljobid);
return ack;
}
bool fire_internal(const std::string& ctljobid, ControlJobEvent event) {
auto* cj = get(ctljobid);
if (!cj) return false;
return cj->fsm->on_internal(event);
bool ok = cj->fsm->on_internal(event);
if (ok && persistent_) write_record_(ctljobid);
return ok;
}
bool remove(const std::string& ctljobid) {
return jobs_.erase(ctljobid) > 0;
auto it = jobs_.find(ctljobid);
if (it == jobs_.end()) return false;
if (persistent_ && !it->second.journal_path.empty()) {
std::error_code ec;
std::filesystem::remove(it->second.journal_path, ec);
}
jobs_.erase(it);
return true;
}
std::size_t size() const { return jobs_.size(); }
@@ -111,10 +117,142 @@ class ControlJobStore {
return out;
}
// ---- Persistence ----------------------------------------------------
// Per-CJ binary record (`<seq>.cj`). CJ ordering across CJs is not
// managed by E94 (each CJ is independent), so no index file is needed.
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() != ".cj") 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->ctljobid, std::move(rec->prjobids), rec->state);
auto* cj = get(rec->ctljobid);
if (cj) cj->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 ctljobid;
std::vector<std::string> prjobids;
ControlJobState state;
};
// CJ record:
// [u8 magic = 0xC8][u8 version = 1][u8 state]
// [u16 ctljobid_len][ctljobid_bytes]
// [u16 pj_count][repeat: u16 len + bytes]
static constexpr uint8_t kMagic = 0xC8;
static constexpr uint8_t kVersion = 0x01;
void install_(const std::string& ctljobid,
std::vector<std::string> prjobids,
ControlJobState restored) {
auto fsm = std::make_unique<ControlJobStateMachine>(factory_(), restored);
const std::string id = ctljobid;
if (on_change_) {
auto cb = on_change_;
fsm->set_state_change_handler(
[cb, id](ControlJobState from, ControlJobState to,
ControlJobEvent trig) { cb(id, from, to, trig); });
}
jobs_.emplace(ctljobid,
ControlJob{ctljobid, std::move(prjobids),
std::move(fsm), {}});
}
void write_record_(const std::string& ctljobid) {
namespace fs = std::filesystem;
auto* cj = get(ctljobid);
if (!cj) return;
if (cj->journal_path.empty()) {
char namebuf[32];
std::snprintf(namebuf, sizeof(namebuf), "%010llu.cj",
static_cast<unsigned long long>(next_seq_++));
cj->journal_path = journal_dir_ / namebuf;
}
fs::path tmp = cj->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>(cj->fsm->state())};
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(cj->ctljobid);
uint16_t pjc = static_cast<uint16_t>(cj->prjobids.size());
uint8_t pb[2] = {static_cast<uint8_t>(pjc >> 8), static_cast<uint8_t>(pjc)};
out.write(reinterpret_cast<const char*>(pb), 2);
for (const auto& id : cj->prjobids) write_str(id);
out.flush();
out.close();
std::error_code ec;
fs::rename(tmp, cj->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<ControlJobState>(header[2]);
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.ctljobid)) return std::nullopt;
uint8_t pb[2];
in.read(reinterpret_cast<char*>(pb), 2);
if (!in) return std::nullopt;
uint16_t pjc = static_cast<uint16_t>((uint16_t(pb[0]) << 8) | pb[1]);
r.prjobids.resize(pjc);
for (auto& id : r.prjobids) {
if (!read_str(id)) return std::nullopt;
}
return r;
}
std::map<std::string, ControlJob> jobs_;
TransitionTableFactory factory_;
StateChangeHandler on_change_;
bool persistent_ = false;
std::filesystem::path journal_dir_;
uint64_t next_seq_ = 0;
};
} // namespace secsgem::gem