From 8a48ffeed43ee65efff2e3abb5dee67048f1fc20 Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Wed, 10 Jun 2026 18:57:53 +0200 Subject: [PATCH] feat(gem): multi-observer state-change handlers via HandlerSlot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The single-slot set_*_handler pattern was a structural blocker, hit twice: the daemon could not observe control-state changes because register_default_handlers owns the slot, forcing GetControlState to read the FSM cross-thread (a data race), and blocking WatchHealth and the Subscribe stream's ControlStateChange variant. HandlerSlot keeps a primary slot with exact legacy semantics (set_ replaces — one existing test depends on replacement) plus an append-only observer list (add_) that survives set_ calls. Fire sites are textually unchanged (operator bool / operator() / assign-from-function). Applied to ControlStateMachine + ProcessJobStore + ControlJobStore (the roadmap-critical three; the remaining single-slot classes follow the same 3-line pattern as needed). EquipmentRuntime gains an atomic control-state mirror registered as an observer — control_state() is now safe from any thread, retiring the GetControlState race — plus add_control_state_observer and add_link_observer (selected/closed fan-out), the hooks WatchHealth and Subscribe need. Tests: observer ordering, set-replaces-primary-but-observers-survive, observers-without-primary, PJ-store coexistence, and the runtime scenario that was previously impossible (mirror + observer + default-handlers set_). Core 464/464 (2816 assertions), daemon 16/16, live GEM300 demo passes with single-fire control-state transitions. Co-Authored-By: Claude Fable 5 --- apps/equipment_service.hpp | 3 +- include/secsgem/gem/control_state.hpp | 7 ++- include/secsgem/gem/handler_slot.hpp | 51 +++++++++++++++++++++ include/secsgem/gem/runtime.hpp | 21 ++++++++- include/secsgem/gem/store/control_jobs.hpp | 4 +- include/secsgem/gem/store/process_jobs.hpp | 7 ++- src/gem/runtime.cpp | 14 +++++- tests/test_control_state.cpp | 53 ++++++++++++++++++++++ tests/test_process_jobs.cpp | 16 +++++++ tests/test_runtime.cpp | 23 ++++++++++ 10 files changed, 192 insertions(+), 7 deletions(-) create mode 100644 include/secsgem/gem/handler_slot.hpp diff --git a/apps/equipment_service.hpp b/apps/equipment_service.hpp index 9664bd1..6557c5a 100644 --- a/apps/equipment_service.hpp +++ b/apps/equipment_service.hpp @@ -132,8 +132,7 @@ class EquipmentService final : public pb::Equipment::Service { grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*, pb::ControlState* resp) override { - // NOTE: reads the FSM state from a gRPC thread — a benign-width race - // documented in docs/DAEMON_ROADMAP.md (fix: atomic state mirror). + // Thread-safe: control_state() reads the runtime's atomic mirror. resp->set_state(to_proto_state(rt_.control_state())); return grpc::Status::OK; } diff --git a/include/secsgem/gem/control_state.hpp b/include/secsgem/gem/control_state.hpp index b040dc7..c0aa648 100644 --- a/include/secsgem/gem/control_state.hpp +++ b/include/secsgem/gem/control_state.hpp @@ -6,6 +6,8 @@ #include #include +#include "secsgem/gem/handler_slot.hpp" + namespace secsgem::gem { // E30 §6.2 Control State Model. @@ -94,7 +96,10 @@ class ControlStateMachine { ControlState state() const { return state_; } bool online() const { return is_online(state_); } + // Replaces the primary handler (legacy single-slot semantics). void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); } + // Appends an observer that survives set_state_change_handler calls. + void add_state_change_handler(StateChangeHandler h) { on_change_.add(std::move(h)); } // Operator actions. Return true if a transition (or self-ack) was found. bool operator_online(); @@ -113,7 +118,7 @@ class ControlStateMachine { ControlTransitionTable table_; ControlState state_; - StateChangeHandler on_change_; + HandlerSlot on_change_; }; } // namespace secsgem::gem diff --git a/include/secsgem/gem/handler_slot.hpp b/include/secsgem/gem/handler_slot.hpp new file mode 100644 index 0000000..29130f9 --- /dev/null +++ b/include/secsgem/gem/handler_slot.hpp @@ -0,0 +1,51 @@ +#pragma once + +#include +#include +#include + +namespace secsgem::gem { + +// Replaces the single-slot `std::function on_change_` pattern with a primary +// slot plus an append-only observer list, so multiple parties can watch one +// state machine without fighting over a single handler. +// +// - `slot = fn;` / set_*_handler(fn) -> sets/replaces the PRIMARY handler. +// Exactly the legacy single-slot semantics: callers that always owned the +// slot (register_default_handlers, apps, tests) keep working unchanged, +// including replacement. +// - `slot.add(fn)` / add_*_handler(fn) -> appends an OBSERVER. Observers are +// immune to later set_ calls — this is what lets the runtime keep an +// atomic control-state mirror (and later: WatchHealth, the Subscribe +// stream) while default_handlers still owns the primary slot. +// +// Invocation order: primary first, then observers in registration order. +// Copyable (fire sites that snapshot the handler before invoking keep +// working). Not thread-safe: register on the owning thread before the +// io_context runs, fire on the io thread — same contract as before. +template +class HandlerSlot { + public: + using Fn = std::function; + + HandlerSlot& operator=(Fn f) { + primary_ = std::move(f); + return *this; + } + + void add(Fn f) { observers_.push_back(std::move(f)); } + + void operator()(Args... args) const { + if (primary_) primary_(args...); + for (const auto& o : observers_) + if (o) o(args...); + } + + explicit operator bool() const { return static_cast(primary_) || !observers_.empty(); } + + private: + Fn primary_; + std::vector observers_; +}; + +} // namespace secsgem::gem diff --git a/include/secsgem/gem/runtime.hpp b/include/secsgem/gem/runtime.hpp index 51005de..236ba4f 100644 --- a/include/secsgem/gem/runtime.hpp +++ b/include/secsgem/gem/runtime.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -8,6 +9,7 @@ #include #include #include +#include #include "secsgem/config/loader.hpp" #include "secsgem/endpoint.hpp" @@ -76,7 +78,22 @@ class EquipmentRuntime { } // ---- control state ------------------------------------------------------- - ControlState control_state() const { return sm_->state(); } + // Safe from any thread: reads an atomic mirror updated by a state-machine + // observer, so gRPC threads never touch the FSM the io thread owns. + ControlState control_state() const { return control_state_cache_.load(std::memory_order_relaxed); } + + // Observe control-state transitions (fires on the io thread). Survives the + // primary set_state_change_handler that register_default_handlers installs. + // Register before run()/run_async(). + void add_control_state_observer(ControlStateMachine::StateChangeHandler h) { + sm_->add_state_change_handler(std::move(h)); + } + + // Observe HSMS link state: fires on the io thread with true when a session + // reaches SELECTED, false when the connection closes. Register before + // run()/run_async(). Foundation for the daemon's WatchHealth stream. + using LinkObserver = std::function; + void add_link_observer(LinkObserver h) { link_observers_.push_back(std::move(h)); } // Deliver a primary to the host, or spool it if there's no SELECTED session. // Call on the io thread (e.g. from a router handler or a posted emitter). @@ -100,6 +117,8 @@ class EquipmentRuntime { config::EquipmentDescriptor descriptor_; std::unique_ptr server_; Router router_; + std::atomic control_state_cache_{ControlState::HostOffline}; + std::vector link_observers_; std::optional> work_; std::thread io_thread_; }; diff --git a/include/secsgem/gem/store/control_jobs.hpp b/include/secsgem/gem/store/control_jobs.hpp index 6ff849b..37ca185 100644 --- a/include/secsgem/gem/store/control_jobs.hpp +++ b/include/secsgem/gem/store/control_jobs.hpp @@ -12,6 +12,7 @@ #include #include +#include "secsgem/gem/handler_slot.hpp" #include "secsgem/gem/control_job_state.hpp" namespace secsgem::gem { @@ -258,7 +259,8 @@ class ControlJobStore { std::map jobs_; TransitionTableFactory factory_; - StateChangeHandler on_change_; + HandlerSlot on_change_; bool persistent_ = false; std::filesystem::path journal_dir_; uint64_t next_seq_ = 0; diff --git a/include/secsgem/gem/store/process_jobs.hpp b/include/secsgem/gem/store/process_jobs.hpp index 729434c..eb973cf 100644 --- a/include/secsgem/gem/store/process_jobs.hpp +++ b/include/secsgem/gem/store/process_jobs.hpp @@ -13,6 +13,7 @@ #include #include +#include "secsgem/gem/handler_slot.hpp" #include "secsgem/gem/e40_constants.hpp" #include "secsgem/gem/process_job_state.hpp" #include "secsgem/secs2/codec.hpp" @@ -72,7 +73,10 @@ class ProcessJobStore { ProcessJobStore& operator=(ProcessJobStore&&) = delete; void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); } + // set_ replaces the primary handler (legacy semantics); add_ appends an + // observer that survives set_ calls (see handler_slot.hpp). void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); } + void add_state_change_handler(StateChangeHandler h) { on_change_.add(std::move(h)); } enum class CreateResult { Created, @@ -512,7 +516,8 @@ class ProcessJobStore { std::map jobs_; std::vector order_; // queue position (E40 HOQ-aware) TransitionTableFactory factory_; - StateChangeHandler on_change_; + HandlerSlot on_change_; bool persistent_ = false; std::filesystem::path journal_dir_; uint64_t next_seq_ = 0; diff --git a/src/gem/runtime.cpp b/src/gem/runtime.cpp index 7fe2fa9..f764fc0 100644 --- a/src/gem/runtime.cpp +++ b/src/gem/runtime.cpp @@ -17,6 +17,14 @@ EquipmentRuntime::EquipmentRuntime(const Config& cfg) descriptor_ = config::load_equipment(cfg.equipment_yaml, *model_); auto sm_cfg = config::load_control_state(cfg.control_state_yaml); sm_ = std::make_shared(sm_cfg.table, sm_cfg.initial); + // Keep an atomic mirror of the control state so control_state() is safe + // from any thread. Registered as an observer (add_), so it survives the + // primary handler register_default_handlers installs later. + control_state_cache_.store(sm_->state(), std::memory_order_relaxed); + sm_->add_state_change_handler( + [this](ControlState, ControlState to, ControlEvent) { + control_state_cache_.store(to, std::memory_order_relaxed); + }); auto pj = config::load_process_job_state(cfg.process_job_yaml); auto cj = config::load_control_job_state(cfg.control_job_yaml); @@ -107,10 +115,14 @@ void EquipmentRuntime::clear_alarm(uint32_t alid) { void EquipmentRuntime::wire_connection() { server_->on_connection([this](std::shared_ptr conn) { *active_conn_ = conn; - conn->set_closed_handler([this](const std::string&) { active_conn_->reset(); }); + conn->set_closed_handler([this](const std::string&) { + active_conn_->reset(); + for (const auto& o : link_observers_) o(false); + }); conn->set_selected_handler([this]() { log(std::string("host is online; control=") + control_state_name(sm_->state())); + for (const auto& o : link_observers_) o(true); if (model_->spool.size() == 0) return; asio::post(io_, [this]() { auto c = active_conn_->lock(); diff --git a/tests/test_control_state.cpp b/tests/test_control_state.cpp index 0fa7a06..c2e93e1 100644 --- a/tests/test_control_state.cpp +++ b/tests/test_control_state.cpp @@ -151,3 +151,56 @@ TEST_CASE("custom table: a row that only sets ack, no transition") { CHECK(sm.on_host_request_online() == OnlineAck::NotAccept); CHECK(sm.state() == ControlState::HostOffline); } + +// ---- multi-observer (HandlerSlot) ------------------------------------------ + +TEST_CASE("add_state_change_handler observers fire alongside the primary, in order") { + ControlStateMachine sm; + std::vector order; + Recorder primary, obs1, obs2; + + sm.set_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) { + order.push_back(0); + primary.handler()(f, t, e); + }); + sm.add_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) { + order.push_back(1); + obs1.handler()(f, t, e); + }); + sm.add_state_change_handler([&](ControlState f, ControlState t, ControlEvent e) { + order.push_back(2); + obs2.handler()(f, t, e); + }); + + CHECK(sm.on_host_request_online() == OnlineAck::Accept); // 2 transitions + + // All three saw both transitions, primary before observers each time. + CHECK(primary.changes.size() == 2); + CHECK(obs1.changes.size() == 2); + CHECK(obs2.changes.size() == 2); + REQUIRE(order.size() == 6); + CHECK(order == std::vector{0, 1, 2, 0, 1, 2}); +} + +TEST_CASE("set_state_change_handler replaces the primary but observers survive") { + ControlStateMachine sm; + Recorder old_primary, new_primary, observer; + + sm.set_state_change_handler(old_primary.handler()); + sm.add_state_change_handler(observer.handler()); + sm.set_state_change_handler(new_primary.handler()); // replaces old_primary ONLY + + sm.on_host_request_online(); + + CHECK(old_primary.changes.empty()); // replaced + CHECK(new_primary.changes.size() == 2); // active primary + CHECK(observer.changes.size() == 2); // survived the re-set +} + +TEST_CASE("observers work with no primary handler installed") { + ControlStateMachine sm; + Recorder observer; + sm.add_state_change_handler(observer.handler()); + sm.on_host_request_online(); + CHECK(observer.changes.size() == 2); +} diff --git a/tests/test_process_jobs.cpp b/tests/test_process_jobs.cpp index 41ea08a..d79113c 100644 --- a/tests/test_process_jobs.cpp +++ b/tests/test_process_jobs.cpp @@ -221,3 +221,19 @@ TEST_CASE("Store: set_alert toggles per-PJ alert flag") { CHECK(store.get("PJ-1")->alert_enabled == false); CHECK_FALSE(store.set_alert("ghost", false)); } + +TEST_CASE("ProcessJobStore: add_state_change_handler observer coexists with primary") { + ProcessJobStore store; + int primary_calls = 0, observer_calls = 0; + store.set_state_change_handler( + [&](const std::string&, ProcessJobState, ProcessJobState, + ProcessJobEvent) { ++primary_calls; }); + store.add_state_change_handler( + [&](const std::string&, ProcessJobState, ProcessJobState, + ProcessJobEvent) { ++observer_calls; }); + + CHECK(store.create("PJ-OBS", "RECIPE-A", {"lot1"}) == + ProcessJobStore::CreateResult::Created); // NoState -> Queued fires both + CHECK(primary_calls == 1); + CHECK(observer_calls == 1); +} diff --git a/tests/test_runtime.cpp b/tests/test_runtime.cpp index 17b553c..b4e2b5a 100644 --- a/tests/test_runtime.cpp +++ b/tests/test_runtime.cpp @@ -64,3 +64,26 @@ TEST_CASE("EquipmentRuntime.on_command registers the behaviour hook on the model CHECK(ran); CHECK(res.ack == gem::HostCmdAck::Accept); } + +TEST_CASE("EquipmentRuntime control-state mirror tracks transitions; observers coexist with default handlers") { + gem::EquipmentRuntime rt(test_config()); + + int observed = 0; + rt.add_control_state_observer( + [&](gem::ControlState, gem::ControlState, gem::ControlEvent) { ++observed; }); + + // Simulate what register_default_handlers does: claim the PRIMARY slot + // after the runtime (mirror) and the observer are already registered. + rt.control().set_state_change_handler( + [](gem::ControlState, gem::ControlState, gem::ControlEvent) {}); + + CHECK(rt.control_state() == gem::ControlState::HostOffline); // mirror initial + + rt.control().on_host_request_online(); // HostOffline -> ... -> OnlineRemote + + // The atomic mirror followed the FSM, and the added observer survived the + // primary set_ call — the exact scenario that used to be impossible. + CHECK(rt.control_state() == gem::ControlState::OnlineRemote); + CHECK(rt.control_state() == rt.control().state()); + CHECK(observed == 2); // two chained transitions +}