raphael
1189ffc994
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>
446 lines
16 KiB
C++
446 lines
16 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/e40_constants.hpp"
|
|
#include "secsgem/gem/process_job_state.hpp"
|
|
#include "secsgem/secs2/item.hpp"
|
|
|
|
namespace secsgem::gem {
|
|
|
|
// One Process Job record. The FSM is heap-allocated through unique_ptr so
|
|
// the per-PJ state-change handler can capture a stable pointer to
|
|
// `this`-style state without invalidation on map rehash.
|
|
//
|
|
// The MF / recipe-method / rcp-vars / process-params fields are the
|
|
// optional E40-0705 §10.2 trailers on S16F11. Simple callers leave
|
|
// them defaulted; tools that actually need recipe-variable tuning or
|
|
// per-job process parameters populate them.
|
|
struct RcpVar {
|
|
std::string name;
|
|
secs2::Item value;
|
|
};
|
|
|
|
struct ProcessParam {
|
|
std::string name;
|
|
secs2::Item value;
|
|
};
|
|
|
|
struct ProcessJob {
|
|
std::string prjobid;
|
|
std::string ppid; // recipe identifier
|
|
std::vector<std::string> mtrloutspec; // material identifiers
|
|
bool alert_enabled = true; // S16F9 alerts on/off
|
|
MaterialFlag mf = MaterialFlag::Substrate;
|
|
ProcessRecipeMethod prrecipemethod = ProcessRecipeMethod::RecipeOnly;
|
|
std::vector<RcpVar> rcpvars;
|
|
std::vector<ProcessParam> prprocessparams;
|
|
std::unique_ptr<ProcessJobStateMachine> fsm;
|
|
std::filesystem::path journal_path;
|
|
};
|
|
|
|
class ProcessJobStore {
|
|
public:
|
|
using TransitionTableFactory =
|
|
std::function<ProcessJobTransitionTable()>;
|
|
using StateChangeHandler =
|
|
std::function<void(const std::string& prjobid,
|
|
ProcessJobState from, ProcessJobState to,
|
|
ProcessJobEvent trigger)>;
|
|
|
|
ProcessJobStore()
|
|
: factory_([] { return ProcessJobTransitionTable::default_table(); }) {}
|
|
|
|
// The per-PJ FSM closes over `this`, so the store must keep a stable
|
|
// address. unique_ptr makes copying impossible anyway; moves would
|
|
// silently dangle the per-PJ lambdas — disallow them explicitly.
|
|
ProcessJobStore(const ProcessJobStore&) = delete;
|
|
ProcessJobStore& operator=(const ProcessJobStore&) = delete;
|
|
ProcessJobStore(ProcessJobStore&&) = delete;
|
|
ProcessJobStore& operator=(ProcessJobStore&&) = 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 CreateResult {
|
|
Created,
|
|
Denied_AlreadyExists,
|
|
Denied_InvalidPpid,
|
|
};
|
|
|
|
// Validate `ppid` against the optional callback; if accepted, create
|
|
// the PJ in Queued and fire the Created -> Queued change handler.
|
|
CreateResult create(std::string prjobid, std::string ppid,
|
|
std::vector<std::string> materials,
|
|
const std::function<bool(const std::string&)>& ppid_exists =
|
|
[](const std::string&) { return true; }) {
|
|
if (jobs_.count(prjobid)) return CreateResult::Denied_AlreadyExists;
|
|
if (!ppid_exists(ppid)) return CreateResult::Denied_InvalidPpid;
|
|
install_(prjobid, std::move(ppid), std::move(materials),
|
|
/*alert=*/true, MaterialFlag::Substrate,
|
|
ProcessRecipeMethod::RecipeOnly, ProcessJobState::Queued);
|
|
order_.push_back(prjobid);
|
|
if (persistent_) {
|
|
write_record_(prjobid);
|
|
write_order_();
|
|
}
|
|
// Synthetic NoState -> Queued so subscribers observe creation.
|
|
if (on_change_) {
|
|
on_change_(jobs_.find(prjobid)->first, ProcessJobState::NoState,
|
|
ProcessJobState::Queued, ProcessJobEvent::Created);
|
|
}
|
|
return CreateResult::Created;
|
|
}
|
|
|
|
// After `create`, populate the optional E40-0705 fields on the new PJ.
|
|
// Returns true if the PJ exists. These fields don't influence the FSM;
|
|
// they're carried so the tool's recipe engine can read them later.
|
|
bool set_e40_extras(const std::string& prjobid,
|
|
MaterialFlag mf,
|
|
ProcessRecipeMethod prrecipemethod,
|
|
std::vector<RcpVar> rcpvars,
|
|
std::vector<ProcessParam> params) {
|
|
auto it = jobs_.find(prjobid);
|
|
if (it == jobs_.end()) return false;
|
|
it->second.mf = mf;
|
|
it->second.prrecipemethod = prrecipemethod;
|
|
it->second.rcpvars = std::move(rcpvars);
|
|
it->second.prprocessparams = std::move(params);
|
|
if (persistent_) write_record_(prjobid);
|
|
return true;
|
|
}
|
|
|
|
bool has(const std::string& prjobid) const {
|
|
return jobs_.count(prjobid) > 0;
|
|
}
|
|
|
|
const ProcessJob* get(const std::string& prjobid) const {
|
|
auto it = jobs_.find(prjobid);
|
|
return it == jobs_.end() ? nullptr : &it->second;
|
|
}
|
|
ProcessJob* get(const std::string& prjobid) {
|
|
auto it = jobs_.find(prjobid);
|
|
return it == jobs_.end() ? nullptr : &it->second;
|
|
}
|
|
|
|
ProcessJobState state(const std::string& prjobid) const {
|
|
auto it = jobs_.find(prjobid);
|
|
return it == jobs_.end() ? ProcessJobState::NoState
|
|
: it->second.fsm->state();
|
|
}
|
|
|
|
// Host-initiated S16F5 PRJobCommand. Returns InvalidObject for unknown
|
|
// PJs; CannotDoNow when the current PJ state has no row for the event.
|
|
// For HOQ the FSM gates legality (Queued only) and the store performs
|
|
// the actual reorder so the next CJ Select picks this PJ first.
|
|
HostCmdAck on_host_command(const std::string& prjobid, ProcessJobEvent event) {
|
|
auto* pj = get(prjobid);
|
|
if (!pj) return HostCmdAck::InvalidObject;
|
|
HostCmdAck ack = pj->fsm->on_host_command(event);
|
|
if (ack == HostCmdAck::Accept) {
|
|
if (event == ProcessJobEvent::HeadOfQueue) {
|
|
move_to_head(prjobid);
|
|
if (persistent_) write_order_();
|
|
} else if (persistent_) {
|
|
write_record_(prjobid);
|
|
}
|
|
}
|
|
return ack;
|
|
}
|
|
|
|
// Position of `prjobid` in the queue order (insertion order; HOQ rewrites
|
|
// it). Returns -1 if unknown. Exposed for tests and CJ promotion logic.
|
|
int position(const std::string& prjobid) const {
|
|
auto it = std::find(order_.begin(), order_.end(), prjobid);
|
|
return it == order_.end() ? -1 : static_cast<int>(it - order_.begin());
|
|
}
|
|
|
|
// Internal events from the application (recipe runner etc.).
|
|
bool fire_internal(const std::string& prjobid, ProcessJobEvent event) {
|
|
auto* pj = get(prjobid);
|
|
if (!pj) return false;
|
|
bool ok = pj->fsm->on_internal(event);
|
|
if (ok && persistent_) write_record_(prjobid);
|
|
return ok;
|
|
}
|
|
|
|
// Dequeue S16F13: only legal while QUEUED.
|
|
HostCmdAck dequeue(const std::string& prjobid) {
|
|
auto it = jobs_.find(prjobid);
|
|
if (it == jobs_.end()) return HostCmdAck::InvalidObject;
|
|
if (it->second.fsm->state() != ProcessJobState::Queued)
|
|
return HostCmdAck::CannotDoNow;
|
|
if (persistent_ && !it->second.journal_path.empty()) {
|
|
std::error_code ec;
|
|
std::filesystem::remove(it->second.journal_path, ec);
|
|
}
|
|
erase_from_order(prjobid);
|
|
jobs_.erase(it);
|
|
if (persistent_) write_order_();
|
|
return HostCmdAck::Accept;
|
|
}
|
|
|
|
// Remove a terminal job from the store (caller responsibility to ensure
|
|
// ProcessComplete before deleting). Used after CJ Completed cleanup.
|
|
bool remove(const std::string& prjobid) {
|
|
auto it = jobs_.find(prjobid);
|
|
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);
|
|
}
|
|
erase_from_order(prjobid);
|
|
jobs_.erase(it);
|
|
if (persistent_) write_order_();
|
|
return true;
|
|
}
|
|
|
|
// Per-PJ S16F9 alert gate. E40 §10.3 leaves PRALERT control to the host
|
|
// (S16F1/F2 in the full multi-create form, which we don't model yet);
|
|
// exposed here so application code can toggle alerts directly.
|
|
bool set_alert(const std::string& prjobid, bool enabled) {
|
|
auto* pj = get(prjobid);
|
|
if (!pj) return false;
|
|
pj->alert_enabled = enabled;
|
|
if (persistent_) write_record_(prjobid);
|
|
return true;
|
|
}
|
|
|
|
std::size_t size() const { return jobs_.size(); }
|
|
|
|
// Insertion-order list of PRJOBIDs, mutated by HOQ. Used by CJ promotion
|
|
// logic and by tests that need to assert ordering.
|
|
const std::vector<std::string>& ids() const { return order_; }
|
|
|
|
// ---- Persistence ----------------------------------------------------
|
|
// One file per PJ (`<seq>.pj`); a single `order.idx` index records the
|
|
// queue ordering (insertion-order modulo HOQ). Replay reads the index
|
|
// first to preserve ordering across restarts. Extras (rcpvars,
|
|
// prprocessparams) are skipped in v1 — they carry secs2::Item variants
|
|
// and aren't necessary for the FSM lifecycle. Callers that need them
|
|
// post-restart should call set_e40_extras() again.
|
|
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::map<std::string, ProcessJob> staging;
|
|
std::map<std::string, fs::path> paths;
|
|
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() != ".pj") 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->prjobid, rec->ppid, std::move(rec->materials),
|
|
rec->alert, rec->mf, rec->prrecipemethod, rec->state);
|
|
auto* pj = get(rec->prjobid);
|
|
if (pj) pj->journal_path = path;
|
|
if (seq >= next_seq_) next_seq_ = seq + 1;
|
|
}
|
|
|
|
// Rebuild order_ from the index file when present; otherwise fall
|
|
// back to creation order (already insertion-sorted by seq above).
|
|
std::vector<std::string> idx = read_order_();
|
|
order_.clear();
|
|
if (!idx.empty()) {
|
|
for (auto& id : idx) if (jobs_.count(id)) order_.push_back(id);
|
|
// Append any PJs not in the index (defensive: shouldn't happen).
|
|
for (auto& kv : jobs_) {
|
|
if (std::find(order_.begin(), order_.end(), kv.first) == order_.end())
|
|
order_.push_back(kv.first);
|
|
}
|
|
} else {
|
|
for (auto& kv : jobs_) order_.push_back(kv.first);
|
|
}
|
|
}
|
|
|
|
bool persistence_enabled() const { return persistent_; }
|
|
std::filesystem::path journal_dir() const { return journal_dir_; }
|
|
|
|
private:
|
|
struct Record {
|
|
std::string prjobid;
|
|
std::string ppid;
|
|
std::vector<std::string> materials;
|
|
bool alert;
|
|
MaterialFlag mf;
|
|
ProcessRecipeMethod prrecipemethod;
|
|
ProcessJobState state;
|
|
};
|
|
|
|
// PJ record:
|
|
// [u8 magic = 0xC7][u8 version = 1]
|
|
// [u8 state][u8 alert][u8 mf][u8 prrecipemethod]
|
|
// [u16 prjobid_len][prjobid_bytes]
|
|
// [u16 ppid_len][ppid_bytes]
|
|
// [u16 material_count][repeat: u16 len + bytes]
|
|
static constexpr uint8_t kMagic = 0xC7;
|
|
static constexpr uint8_t kVersion = 0x01;
|
|
|
|
void install_(const std::string& prjobid, std::string ppid,
|
|
std::vector<std::string> materials, bool alert,
|
|
MaterialFlag mf, ProcessRecipeMethod prrecipemethod,
|
|
ProcessJobState restored) {
|
|
auto fsm = std::make_unique<ProcessJobStateMachine>(factory_(), restored);
|
|
fsm->set_state_change_handler(
|
|
[this, id = prjobid](ProcessJobState from, ProcessJobState to,
|
|
ProcessJobEvent trig) {
|
|
if (on_change_) on_change_(id, from, to, trig);
|
|
});
|
|
ProcessJob pj;
|
|
pj.prjobid = prjobid;
|
|
pj.ppid = std::move(ppid);
|
|
pj.mtrloutspec = std::move(materials);
|
|
pj.alert_enabled = alert;
|
|
pj.mf = mf;
|
|
pj.prrecipemethod = prrecipemethod;
|
|
pj.fsm = std::move(fsm);
|
|
jobs_.emplace(prjobid, std::move(pj));
|
|
}
|
|
|
|
void write_record_(const std::string& prjobid) {
|
|
namespace fs = std::filesystem;
|
|
auto* pj = get(prjobid);
|
|
if (!pj) return;
|
|
if (pj->journal_path.empty()) {
|
|
char namebuf[32];
|
|
std::snprintf(namebuf, sizeof(namebuf), "%010llu.pj",
|
|
static_cast<unsigned long long>(next_seq_++));
|
|
pj->journal_path = journal_dir_ / namebuf;
|
|
}
|
|
fs::path tmp = pj->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>(pj->fsm->state()),
|
|
static_cast<uint8_t>(pj->alert_enabled ? 1 : 0),
|
|
static_cast<uint8_t>(pj->mf),
|
|
static_cast<uint8_t>(pj->prrecipemethod)};
|
|
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(pj->prjobid);
|
|
write_str(pj->ppid);
|
|
uint16_t mc = static_cast<uint16_t>(pj->mtrloutspec.size());
|
|
uint8_t mb[2] = {static_cast<uint8_t>(mc >> 8), static_cast<uint8_t>(mc)};
|
|
out.write(reinterpret_cast<const char*>(mb), 2);
|
|
for (const auto& m : pj->mtrloutspec) write_str(m);
|
|
out.flush();
|
|
out.close();
|
|
std::error_code ec;
|
|
fs::rename(tmp, pj->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.state = static_cast<ProcessJobState>(header[2]);
|
|
r.alert = header[3] != 0;
|
|
r.mf = static_cast<MaterialFlag>(header[4]);
|
|
r.prrecipemethod = static_cast<ProcessRecipeMethod>(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.prjobid)) return std::nullopt;
|
|
if (!read_str(r.ppid)) return std::nullopt;
|
|
uint8_t mb[2];
|
|
in.read(reinterpret_cast<char*>(mb), 2);
|
|
if (!in) return std::nullopt;
|
|
uint16_t mc = static_cast<uint16_t>((uint16_t(mb[0]) << 8) | mb[1]);
|
|
r.materials.resize(mc);
|
|
for (auto& m : r.materials) {
|
|
if (!read_str(m)) return std::nullopt;
|
|
}
|
|
return r;
|
|
}
|
|
|
|
void write_order_() {
|
|
namespace fs = std::filesystem;
|
|
fs::path target = journal_dir_ / "order.idx";
|
|
fs::path tmp = target;
|
|
tmp += ".tmp";
|
|
std::ofstream out(tmp, std::ios::binary | std::ios::trunc);
|
|
if (!out) return;
|
|
for (const auto& id : order_) {
|
|
out << id << '\n';
|
|
}
|
|
out.flush();
|
|
out.close();
|
|
std::error_code ec;
|
|
fs::rename(tmp, target, ec);
|
|
if (ec) fs::remove(tmp, ec);
|
|
}
|
|
|
|
std::vector<std::string> read_order_() const {
|
|
std::vector<std::string> out;
|
|
std::ifstream in(journal_dir_ / "order.idx");
|
|
if (!in) return out;
|
|
std::string line;
|
|
while (std::getline(in, line)) {
|
|
if (!line.empty()) out.push_back(line);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
void erase_from_order(const std::string& prjobid) {
|
|
auto it = std::find(order_.begin(), order_.end(), prjobid);
|
|
if (it != order_.end()) order_.erase(it);
|
|
}
|
|
void move_to_head(const std::string& prjobid) {
|
|
auto it = std::find(order_.begin(), order_.end(), prjobid);
|
|
if (it == order_.end() || it == order_.begin()) return;
|
|
std::rotate(order_.begin(), it, it + 1);
|
|
}
|
|
|
|
std::map<std::string, ProcessJob> jobs_;
|
|
std::vector<std::string> order_; // queue position (E40 HOQ-aware)
|
|
TransitionTableFactory factory_;
|
|
StateChangeHandler on_change_;
|
|
bool persistent_ = false;
|
|
std::filesystem::path journal_dir_;
|
|
uint64_t next_seq_ = 0;
|
|
};
|
|
|
|
} // namespace secsgem::gem
|