From 1189ffc994de38dca441dc78aa3fd6f51cfb6d58 Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Tue, 9 Jun 2026 10:35:26 +0200 Subject: [PATCH] persistence: ProcessJobStore + ControlJobStore enable_persistence(dir) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- CMakeLists.txt | 1 + include/secsgem/gem/store/control_jobs.hpp | 172 +++++++++++-- include/secsgem/gem/store/process_jobs.hpp | 266 +++++++++++++++++++-- tests/test_job_persistence.cpp | 172 +++++++++++++ 4 files changed, 570 insertions(+), 41 deletions(-) create mode 100644 tests/test_job_persistence.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 53cd7b5..ed0658b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -108,6 +108,7 @@ add_executable(secsgem_tests tests/test_loader.cpp tests/test_process_jobs.cpp tests/test_control_jobs.cpp + tests/test_job_persistence.cpp tests/test_exceptions.cpp tests/test_carrier_state.cpp tests/test_carriers.cpp diff --git a/include/secsgem/gem/store/control_jobs.hpp b/include/secsgem/gem/store/control_jobs.hpp index 938c2a5..2661d11 100644 --- a/include/secsgem/gem/store/control_jobs.hpp +++ b/include/secsgem/gem/store/control_jobs.hpp @@ -1,9 +1,14 @@ #pragma once +#include +#include +#include +#include #include #include #include #include +#include #include #include @@ -16,6 +21,7 @@ struct ControlJob { std::string ctljobid; std::vector prjobids; std::unique_ptr 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(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 (`.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> 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 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 prjobids, + ControlJobState restored) { + auto fsm = std::make_unique(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(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(cj->fsm->state())}; + out.write(reinterpret_cast(header), sizeof(header)); + auto write_str = [&](const std::string& str) { + uint16_t n = static_cast(str.size()); + uint8_t lb[2] = {static_cast(n >> 8), static_cast(n)}; + out.write(reinterpret_cast(lb), 2); + out.write(str.data(), static_cast(n)); + }; + write_str(cj->ctljobid); + uint16_t pjc = static_cast(cj->prjobids.size()); + uint8_t pb[2] = {static_cast(pjc >> 8), static_cast(pjc)}; + out.write(reinterpret_cast(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 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(header), sizeof(header)); + if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt; + Record r; + r.state = static_cast(header[2]); + auto read_str = [&](std::string& out) { + uint8_t lb[2]; + in.read(reinterpret_cast(lb), 2); + if (!in) return false; + uint16_t n = static_cast((uint16_t(lb[0]) << 8) | lb[1]); + out.resize(n); + in.read(out.data(), n); + return static_cast(in) || n == 0; + }; + if (!read_str(r.ctljobid)) return std::nullopt; + uint8_t pb[2]; + in.read(reinterpret_cast(pb), 2); + if (!in) return std::nullopt; + uint16_t pjc = static_cast((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 jobs_; TransitionTableFactory factory_; StateChangeHandler on_change_; + bool persistent_ = false; + std::filesystem::path journal_dir_; + uint64_t next_seq_ = 0; }; } // namespace secsgem::gem diff --git a/include/secsgem/gem/store/process_jobs.hpp b/include/secsgem/gem/store/process_jobs.hpp index da91d13..a4bfc0e 100644 --- a/include/secsgem/gem/store/process_jobs.hpp +++ b/include/secsgem/gem/store/process_jobs.hpp @@ -1,10 +1,15 @@ #pragma once #include +#include +#include +#include +#include #include #include #include #include +#include #include #include @@ -42,6 +47,7 @@ struct ProcessJob { std::vector rcpvars; std::vector prprocessparams; std::unique_ptr fsm; + std::filesystem::path journal_path; }; class ProcessJobStore { @@ -81,27 +87,15 @@ class ProcessJobStore { [](const std::string&) { return true; }) { if (jobs_.count(prjobid)) return CreateResult::Denied_AlreadyExists; if (!ppid_exists(ppid)) return CreateResult::Denied_InvalidPpid; - auto fsm = std::make_unique(factory_(), - ProcessJobState::Queued); - // Dispatch through *this so a later set_state_change_handler() takes - // effect for existing PJs (the captured-at-create snapshot pattern - // would otherwise pin the old handler on jobs already in the map). - fsm->set_state_change_handler( - [this, id = prjobid](ProcessJobState from, ProcessJobState to, - ProcessJobEvent trig) { - if (on_change_) on_change_(id, from, to, trig); - }); + install_(prjobid, std::move(ppid), std::move(materials), + /*alert=*/true, MaterialFlag::Substrate, + ProcessRecipeMethod::RecipeOnly, ProcessJobState::Queued); order_.push_back(prjobid); - ProcessJob pj; - pj.prjobid = prjobid; - pj.ppid = std::move(ppid); - pj.mtrloutspec = std::move(materials); - pj.alert_enabled = true; - pj.fsm = std::move(fsm); - jobs_.emplace(prjobid, std::move(pj)); - // Synthetic NoState -> Queued so subscribers observe creation. The - // server filters this so it doesn't emit a bogus S16F9 for a PJ - // that's still being acked. + 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); @@ -123,6 +117,7 @@ class ProcessJobStore { it->second.prrecipemethod = prrecipemethod; it->second.rcpvars = std::move(rcpvars); it->second.prprocessparams = std::move(params); + if (persistent_) write_record_(prjobid); return true; } @@ -153,8 +148,13 @@ class ProcessJobStore { auto* pj = get(prjobid); if (!pj) return HostCmdAck::InvalidObject; HostCmdAck ack = pj->fsm->on_host_command(event); - if (ack == HostCmdAck::Accept && event == ProcessJobEvent::HeadOfQueue) { - move_to_head(prjobid); + if (ack == HostCmdAck::Accept) { + if (event == ProcessJobEvent::HeadOfQueue) { + move_to_head(prjobid); + if (persistent_) write_order_(); + } else if (persistent_) { + write_record_(prjobid); + } } return ack; } @@ -170,7 +170,9 @@ class ProcessJobStore { bool fire_internal(const std::string& prjobid, ProcessJobEvent event) { auto* pj = get(prjobid); if (!pj) return false; - return pj->fsm->on_internal(event); + bool ok = pj->fsm->on_internal(event); + if (ok && persistent_) write_record_(prjobid); + return ok; } // Dequeue S16F13: only legal while QUEUED. @@ -179,16 +181,29 @@ class ProcessJobStore { 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); - return jobs_.erase(prjobid) > 0; + jobs_.erase(it); + if (persistent_) write_order_(); + return true; } // Per-PJ S16F9 alert gate. E40 §10.3 leaves PRALERT control to the host @@ -198,6 +213,7 @@ class ProcessJobStore { auto* pj = get(prjobid); if (!pj) return false; pj->alert_enabled = enabled; + if (persistent_) write_record_(prjobid); return true; } @@ -207,7 +223,206 @@ class ProcessJobStore { // logic and by tests that need to assert ordering. const std::vector& ids() const { return order_; } + // ---- 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. + 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 staging; + std::map paths; + std::vector> 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 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 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 materials, bool alert, + MaterialFlag mf, ProcessRecipeMethod prrecipemethod, + ProcessJobState restored) { + auto fsm = std::make_unique(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(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(pj->fsm->state()), + static_cast(pj->alert_enabled ? 1 : 0), + static_cast(pj->mf), + static_cast(pj->prrecipemethod)}; + out.write(reinterpret_cast(header), sizeof(header)); + auto write_str = [&](const std::string& str) { + uint16_t n = static_cast(str.size()); + uint8_t lb[2] = {static_cast(n >> 8), static_cast(n)}; + out.write(reinterpret_cast(lb), 2); + out.write(str.data(), static_cast(n)); + }; + 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); + 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 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(header), sizeof(header)); + if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt; + Record r; + r.state = static_cast(header[2]); + r.alert = header[3] != 0; + r.mf = static_cast(header[4]); + r.prrecipemethod = static_cast(header[5]); + auto read_str = [&](std::string& out) { + uint8_t lb[2]; + in.read(reinterpret_cast(lb), 2); + if (!in) return false; + uint16_t n = static_cast((uint16_t(lb[0]) << 8) | lb[1]); + out.resize(n); + in.read(out.data(), n); + return static_cast(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(mb), 2); + if (!in) return std::nullopt; + uint16_t mc = static_cast((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 read_order_() const { + std::vector 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); @@ -222,6 +437,9 @@ class ProcessJobStore { std::vector 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 diff --git a/tests/test_job_persistence.cpp b/tests/test_job_persistence.cpp new file mode 100644 index 0000000..b256faf --- /dev/null +++ b/tests/test_job_persistence.cpp @@ -0,0 +1,172 @@ +// Persistence tests for ProcessJobStore + ControlJobStore. + +#include + +#include +#include +#include +#include + +#include "secsgem/gem/store/control_jobs.hpp" +#include "secsgem/gem/store/process_jobs.hpp" + +using namespace secsgem::gem; +namespace fs = std::filesystem; + +namespace { +fs::path scratch_dir(const char* tag) { + std::random_device rd; + auto p = fs::temp_directory_path() / + (std::string("secsgem-job-") + tag + "-" + std::to_string(rd())); + fs::remove_all(p); + fs::create_directories(p); + return p; +} +std::size_t count_with_ext(const fs::path& dir, const std::string& ext) { + std::size_t n = 0; + std::error_code ec; + for (auto& e : fs::directory_iterator(dir, ec)) { + if (e.is_regular_file() && e.path().extension() == ext) ++n; + } + return n; +} +} // namespace + +TEST_CASE("ProcessJobStore persistence: state + ordering replay") { + auto dir = scratch_dir("pj-replay"); + { + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("PJ-A", "RCP-1", {"W1", "W2"}) == + ProcessJobStore::CreateResult::Created); + REQUIRE(s.create("PJ-B", "RCP-1", {"W3"}) == + ProcessJobStore::CreateResult::Created); + REQUIRE(s.create("PJ-C", "RCP-2", {"W4", "W5"}) == + ProcessJobStore::CreateResult::Created); + + // Drive PJ-A into Processing. + REQUIRE(s.fire_internal("PJ-A", ProcessJobEvent::Select)); + REQUIRE(s.fire_internal("PJ-A", ProcessJobEvent::SetupComplete)); + REQUIRE(s.on_host_command("PJ-A", ProcessJobEvent::Start) == + HostCmdAck::Accept); + + // HOQ-promote PJ-C to head. + REQUIRE(s.on_host_command("PJ-C", ProcessJobEvent::HeadOfQueue) == + HostCmdAck::Accept); + + REQUIRE(s.set_alert("PJ-B", false)); + + CHECK(count_with_ext(dir, ".pj") == 3); + CHECK(s.position("PJ-C") == 0); + CHECK(s.position("PJ-A") == 1); + CHECK(s.position("PJ-B") == 2); + } + { + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.has("PJ-A")); + REQUIRE(s.has("PJ-B")); + REQUIRE(s.has("PJ-C")); + + // State survived. + CHECK(s.get("PJ-A")->fsm->state() == ProcessJobState::Processing); + CHECK(s.get("PJ-B")->fsm->state() == ProcessJobState::Queued); + CHECK(s.get("PJ-C")->fsm->state() == ProcessJobState::Queued); + + // Recipe + materials survived. + CHECK(s.get("PJ-A")->ppid == "RCP-1"); + CHECK(s.get("PJ-A")->mtrloutspec == std::vector{"W1", "W2"}); + + // Alert flag survived. + CHECK_FALSE(s.get("PJ-B")->alert_enabled); + + // HOQ ordering survived. + CHECK(s.position("PJ-C") == 0); + CHECK(s.position("PJ-A") == 1); + CHECK(s.position("PJ-B") == 2); + } + fs::remove_all(dir); +} + +TEST_CASE("ProcessJobStore persistence: dequeue removes journal + index entry") { + auto dir = scratch_dir("pj-dequeue"); + { + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("PJ-1", "R", {}) == ProcessJobStore::CreateResult::Created); + REQUIRE(s.create("PJ-2", "R", {}) == ProcessJobStore::CreateResult::Created); + CHECK(count_with_ext(dir, ".pj") == 2); + REQUIRE(s.dequeue("PJ-1") == HostCmdAck::Accept); + CHECK(count_with_ext(dir, ".pj") == 1); + } + { + ProcessJobStore s; + s.enable_persistence(dir); + CHECK_FALSE(s.has("PJ-1")); + CHECK(s.has("PJ-2")); + CHECK(s.ids() == std::vector{"PJ-2"}); + } + fs::remove_all(dir); +} + +TEST_CASE("ProcessJobStore persistence: corrupt record dropped, others survive") { + auto dir = scratch_dir("pj-corrupt"); + { + ProcessJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("PJ-OK", "R", {}) == ProcessJobStore::CreateResult::Created); + } + { + std::ofstream f(dir / "9999999999.pj", std::ios::binary); + const char junk[] = {0x00, 0x01}; + f.write(junk, sizeof(junk)); + } + ProcessJobStore s; + s.enable_persistence(dir); + CHECK(s.has("PJ-OK")); + CHECK(s.size() == 1); + fs::remove_all(dir); +} + +TEST_CASE("ControlJobStore persistence: state + prjob list replay") { + auto dir = scratch_dir("cj-replay"); + { + ControlJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("CJ-A", {"PJ-1", "PJ-2"}) == + ControlJobStore::CreateResult::Created); + REQUIRE(s.create("CJ-B", {"PJ-3"}) == + ControlJobStore::CreateResult::Created); + + REQUIRE(s.fire_internal("CJ-A", ControlJobEvent::Select)); + REQUIRE(s.fire_internal("CJ-A", ControlJobEvent::SetupComplete)); + REQUIRE(s.on_host_command("CJ-A", ControlJobEvent::Start) == + HostCmdAck::Accept); + + CHECK(count_with_ext(dir, ".cj") == 2); + } + { + ControlJobStore s; + s.enable_persistence(dir); + REQUIRE(s.has("CJ-A")); + REQUIRE(s.has("CJ-B")); + CHECK(s.get("CJ-A")->fsm->state() == ControlJobState::Executing); + CHECK(s.get("CJ-A")->prjobids == std::vector{"PJ-1", "PJ-2"}); + CHECK(s.get("CJ-B")->fsm->state() == ControlJobState::Queued); + } + fs::remove_all(dir); +} + +TEST_CASE("ControlJobStore persistence: remove deletes journal file") { + auto dir = scratch_dir("cj-remove"); + ControlJobStore s; + s.enable_persistence(dir); + REQUIRE(s.create("CJ-X", {"PJ-1"}) == + ControlJobStore::CreateResult::Created); + REQUIRE(s.create("CJ-Y", {"PJ-2"}) == + ControlJobStore::CreateResult::Created); + CHECK(count_with_ext(dir, ".cj") == 2); + REQUIRE(s.remove("CJ-X")); + CHECK(count_with_ext(dir, ".cj") == 1); + fs::remove_all(dir); +}