diff --git a/include/secsgem/gem/store/process_jobs.hpp b/include/secsgem/gem/store/process_jobs.hpp index a4bfc0e..729434c 100644 --- a/include/secsgem/gem/store/process_jobs.hpp +++ b/include/secsgem/gem/store/process_jobs.hpp @@ -15,6 +15,7 @@ #include "secsgem/gem/e40_constants.hpp" #include "secsgem/gem/process_job_state.hpp" +#include "secsgem/secs2/codec.hpp" #include "secsgem/secs2/item.hpp" namespace secsgem::gem { @@ -226,10 +227,10 @@ class ProcessJobStore { // ---- Persistence ---------------------------------------------------- // One file per PJ (`.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. + // first to preserve ordering across restarts. Version 2 of the record + // (current) journals the E40 extras (rcpvars + prprocessparams) by + // round-tripping each value through secs2::encode/decode. Version 1 + // records are accepted on replay; their extras come back empty. void enable_persistence(std::filesystem::path dir) { namespace fs = std::filesystem; journal_dir_ = std::move(dir); @@ -258,7 +259,11 @@ class ProcessJobStore { 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 (pj) { + pj->journal_path = path; + pj->rcpvars = std::move(rec->rcpvars); + pj->prprocessparams = std::move(rec->prprocessparams); + } if (seq >= next_seq_) next_seq_ = seq + 1; } @@ -290,16 +295,23 @@ class ProcessJobStore { MaterialFlag mf; ProcessRecipeMethod prrecipemethod; ProcessJobState state; + std::vector rcpvars; + std::vector prprocessparams; }; - // PJ record: - // [u8 magic = 0xC7][u8 version = 1] + // PJ record (version 2): + // [u8 magic = 0xC7][u8 version] // [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] + // --- v2 additions --- + // [u16 rcpvar_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)] + // [u16 ppparam_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)] + // v1 records (no extras) are accepted on replay; we just skip the + // trailers and the in-memory rcpvars/prprocessparams stay empty. static constexpr uint8_t kMagic = 0xC7; - static constexpr uint8_t kVersion = 0x01; + static constexpr uint8_t kVersion = 0x02; void install_(const std::string& prjobid, std::string ppid, std::vector materials, bool alert, @@ -351,10 +363,30 @@ class ProcessJobStore { }; write_str(pj->prjobid); write_str(pj->ppid); - uint16_t mc = static_cast(pj->mtrloutspec.size()); - uint8_t mb[2] = {static_cast(mc >> 8), static_cast(mc)}; - out.write(reinterpret_cast(mb), 2); + auto write_be16 = [&](uint16_t v) { + uint8_t b[2] = {static_cast(v >> 8), static_cast(v)}; + out.write(reinterpret_cast(b), 2); + }; + auto write_be32 = [&](uint32_t v) { + uint8_t b[4] = {static_cast((v >> 24) & 0xFF), + static_cast((v >> 16) & 0xFF), + static_cast((v >> 8) & 0xFF), + static_cast(v & 0xFF)}; + out.write(reinterpret_cast(b), 4); + }; + write_be16(static_cast(pj->mtrloutspec.size())); for (const auto& m : pj->mtrloutspec) write_str(m); + auto write_named = [&](const std::string& name, const secs2::Item& value) { + write_str(name); + auto enc = secs2::encode(value); + write_be32(static_cast(enc.size())); + out.write(reinterpret_cast(enc.data()), + static_cast(enc.size())); + }; + write_be16(static_cast(pj->rcpvars.size())); + for (const auto& v : pj->rcpvars) write_named(v.name, v.value); + write_be16(static_cast(pj->prprocessparams.size())); + for (const auto& v : pj->prprocessparams) write_named(v.name, v.value); out.flush(); out.close(); std::error_code ec; @@ -367,7 +399,10 @@ class ProcessJobStore { if (!in) return std::nullopt; uint8_t header[6]; in.read(reinterpret_cast(header), sizeof(header)); - if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt; + if (!in || header[0] != kMagic) return std::nullopt; + // v1 and v2 share the same prefix; v2 adds rcpvars+prprocessparams. + if (header[1] != 0x01 && header[1] != 0x02) return std::nullopt; + const uint8_t version = header[1]; Record r; r.state = static_cast(header[2]); r.alert = header[3] != 0; @@ -382,16 +417,57 @@ class ProcessJobStore { in.read(out.data(), n); return static_cast(in) || n == 0; }; + auto read_be16 = [&](uint16_t& v) { + uint8_t b[2]; + in.read(reinterpret_cast(b), 2); + if (!in) return false; + v = static_cast((uint16_t(b[0]) << 8) | b[1]); + return true; + }; + auto read_be32 = [&](uint32_t& v) { + uint8_t b[4]; + in.read(reinterpret_cast(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; + }; if (!read_str(r.prjobid)) return std::nullopt; if (!read_str(r.ppid)) return std::nullopt; - uint8_t mb[2]; - in.read(reinterpret_cast(mb), 2); - if (!in) return std::nullopt; - uint16_t mc = static_cast((uint16_t(mb[0]) << 8) | mb[1]); + uint16_t mc = 0; + if (!read_be16(mc)) return std::nullopt; r.materials.resize(mc); for (auto& m : r.materials) { if (!read_str(m)) return std::nullopt; } + if (version >= 2) { + auto read_named_list = [&](std::vector& dst) -> bool { + uint16_t n = 0; + if (!read_be16(n)) return false; + dst.resize(n); + for (auto& kv : dst) { + if (!read_str(kv.name)) return false; + uint32_t enc_len = 0; + if (!read_be32(enc_len)) return false; + std::vector bytes(enc_len); + if (enc_len) { + in.read(reinterpret_cast(bytes.data()), enc_len); + if (!in) return false; + } + try { kv.value = secs2::decode(bytes); } + catch (...) { return false; } + } + return true; + }; + if (!read_named_list(r.rcpvars)) return std::nullopt; + // ProcessParam has same shape as RcpVar; reuse via reinterpret-style: + std::vector pp_tmp; + if (!read_named_list(pp_tmp)) return std::nullopt; + r.prprocessparams.reserve(pp_tmp.size()); + for (auto& kv : pp_tmp) { + r.prprocessparams.push_back({std::move(kv.name), std::move(kv.value)}); + } + } return r; } diff --git a/tests/test_job_persistence.cpp b/tests/test_job_persistence.cpp index b256faf..9bd65e7 100644 --- a/tests/test_job_persistence.cpp +++ b/tests/test_job_persistence.cpp @@ -157,6 +157,76 @@ TEST_CASE("ControlJobStore persistence: state + prjob list replay") { fs::remove_all(dir); } +TEST_CASE("ProcessJobStore persistence: rcpvars + prprocessparams survive restart") { + auto dir = scratch_dir("pj-extras"); + { + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("PJ-X", "RCP-1", {"W1"}) == + ProcessJobStore::CreateResult::Created); + std::vector rcpvars = { + {"setpoint_c", secsgem::secs2::Item::f4(350.5f)}, + {"chamber_name", secsgem::secs2::Item::ascii("CHAMBER-A")}, + {"flags", secsgem::secs2::Item::u4(uint32_t{0xDEADBEEFu})}, + }; + std::vector params = { + {"recipe_var", secsgem::secs2::Item::list({ + secsgem::secs2::Item::ascii("nested"), + secsgem::secs2::Item::u2(uint16_t{42})})}, + }; + REQUIRE(s.set_e40_extras("PJ-X", MaterialFlag::Substrate, + ProcessRecipeMethod::RecipeOnly, + rcpvars, params)); + } + { + ProcessJobStore s; + s.enable_persistence(dir); + const auto* pj = s.get("PJ-X"); + REQUIRE(pj); + REQUIRE(pj->rcpvars.size() == 3); + CHECK(pj->rcpvars[0].name == "setpoint_c"); + CHECK(pj->rcpvars[0].value.format() == secsgem::secs2::Format::F4); + CHECK(pj->rcpvars[1].value.as_ascii() == "CHAMBER-A"); + CHECK(pj->rcpvars[2].name == "flags"); + + REQUIRE(pj->prprocessparams.size() == 1); + CHECK(pj->prprocessparams[0].name == "recipe_var"); + CHECK(pj->prprocessparams[0].value.is_list()); + CHECK(pj->prprocessparams[0].value.as_list().size() == 2); + CHECK(pj->prprocessparams[0].value.as_list()[0].as_ascii() == "nested"); + } + fs::remove_all(dir); +} + +TEST_CASE("ProcessJobStore persistence: v1 record (no extras) still loads") { + auto dir = scratch_dir("pj-v1"); + // Hand-craft a v1 record on disk. + { + std::ofstream f(dir / "0000000001.pj", std::ios::binary); + // [magic=0xC7][version=1][state=Queued=0][alert=1][mf=0][prrecipemethod=0] + const uint8_t header[6] = {0xC7, 0x01, 0x00, 0x01, 0x00, 0x00}; + f.write(reinterpret_cast(header), 6); + auto write_str = [&](const std::string& s) { + uint16_t n = static_cast(s.size()); + uint8_t lb[2] = {static_cast(n >> 8), static_cast(n)}; + f.write(reinterpret_cast(lb), 2); + f.write(s.data(), n); + }; + write_str("PJ-LEGACY"); // prjobid + write_str("RCP-OLD"); // ppid + const uint8_t mc[2] = {0x00, 0x00}; + f.write(reinterpret_cast(mc), 2); + // No v2 trailer. + } + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.has("PJ-LEGACY")); + CHECK(s.get("PJ-LEGACY")->ppid == "RCP-OLD"); + CHECK(s.get("PJ-LEGACY")->rcpvars.empty()); + CHECK(s.get("PJ-LEGACY")->prprocessparams.empty()); + fs::remove_all(dir); +} + TEST_CASE("ControlJobStore persistence: remove deletes journal file") { auto dir = scratch_dir("cj-remove"); ControlJobStore s;