From 9876dd9b5a4d699c55c99e9523cfac00ef9c13be Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Fri, 26 Jun 2026 15:05:13 +0200 Subject: [PATCH] feat(daemon): D10 carriers + E16 ops RPCs + stress test + virtual fab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the daemon's GEM300 surface and adds two new test tiers. D10 — E87 carriers: CarrierStore gains the HandlerSlot observer pattern (add_id/slot_map/access_handler). The daemon's id-observer forwards host S3F17 decisions onto the Subscribe stream as CarrierAction (PROCEED on a Confirmed transition, CANCEL on CancelCarrier); ReportCarrier drives the flow tool-side: WAITING creates the carrier + records the slot map, IN_ACCESS/COMPLETE advance the access FSM (INVALID_OBJECT on unknown, CANNOT_DO_NOW on an illegal transition). E16 — operations RPCs: Describe (full name inventory: variables/events/ alarms/commands/constants + device header), FlushSpool (purge or drain), SendTerminalMessage (S10F1 tool->host, honest CANNOT_DO_NOW when no host and stream 10 isn't spoolable). Stream responsiveness: Subscribe/WatchHealth poll at 100ms (was 500ms) so a cancelled stream frees its sync-server worker thread promptly — this was found by the new stress test, which hung under Subscribe churn at 500ms. Tests: - A randomized concurrent RPC stress case: 4 threads x 250 seeded ops (set/get/fire/alarm/control-state/describe + Subscribe churn), asserts no failed RPC and a still-responsive engine afterward; prints its seed; a strong TSan target. - A virtual fab (interop/virtual_fab.py + the `fab` compose service / tools/spawn_fab.sh): N daemons, each with a secsgem-py host AND a secsgem_client tool, driven by seeded random traffic with end-to-end invariant checks (set/get round-trips, event->S6F11 and alarm->S5F1 delivery, command->tool->completion). Verified green at N=3 (~150 ops/eq, all commands round-tripped, 0 violations). Wired into run_interop.sh (now 13 steps). Also fixes the CI break from the previous commit: the Python-client lane's test_values.py step lacked PYTHONPATH=clients/python (now step-level env). Two bugs found and fixed while building this, both mine from this batch: 1. carrier test hung on a CancelCarrier of a still-NotConfirmed carrier — a self-transition the FSM doesn't signal, so the observer never fired and the stream Read blocked forever. Fixed to cancel a Confirmed carrier; the NotConfirmed edge is documented as a known E87 limitation. 2. the 500ms stream poll above. Daemon suite 7 cases / 214 assertions; core 475 / 3097; virtual fab green. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitea/workflows/ci.yml | 2 + docker-compose.yml | 13 + docs/DAEMON_ROADMAP.md | 31 ++- include/secsgem/daemon/equipment_service.hpp | 124 +++++++++- include/secsgem/gem/store/carriers.hpp | 16 +- interop/virtual_fab.py | 209 ++++++++++++++++ proto/secsgem/v1/equipment.proto | 41 ++- tests/test_daemon_service.cpp | 248 +++++++++++++++++++ tools/run_interop.sh | 12 +- tools/spawn_fab.sh | 20 ++ 10 files changed, 703 insertions(+), 13 deletions(-) create mode 100644 interop/virtual_fab.py create mode 100755 tools/spawn_fab.sh diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 1bd8097..a966b6d 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -280,6 +280,8 @@ jobs: exit $rc - name: Python client package vs secs_gemd + env: + PYTHONPATH: clients/python run: | /tmp/venv/bin/python clients/python/tests/test_values.py build/secs_gemd --port 5005 --grpc 127.0.0.1:50052 & diff --git a/docker-compose.yml b/docker-compose.yml index 1d60a6e..a3dd975 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -73,6 +73,19 @@ services: - /app/data networks: [secs] + # A virtual fab: FAB_N secs_gemd equipment in one container (HSMS 5100+i, + # gRPC 51000+i). interop/virtual_fab.py attaches a secsgem-py host AND a + # secsgem_client tool to every one and drives seeded random traffic. + fab: + <<: *base + depends_on: + builder: + condition: service_completed_successfully + environment: + FAB_N: "${FAB_N:-3}" + command: ["bash", "tools/spawn_fab.sh"] + networks: [secs] + # Python container preloaded with secsgem-py 0.3.0 for cross-validation # of our C++ HSMS/SECS-II/GEM implementation against the reference library. interop: diff --git a/docs/DAEMON_ROADMAP.md b/docs/DAEMON_ROADMAP.md index 80db4e3..7a6781e 100644 --- a/docs/DAEMON_ROADMAP.md +++ b/docs/DAEMON_ROADMAP.md @@ -211,10 +211,15 @@ debts tax every later phase, and the most valuable tests aren't automated. →Aborting=ABORT; carries recipe + material bindings) + `ReportProcessJob` (SETTING_UP→SetupComplete, COMPLETE→ProcessComplete, ABORTED→AbortComplete; PROCESSING informational; INVALID_OBJECT / - CANNOT_DO_NOW on unknown job / illegal transition). ⬜ Carriers deferred: - `CarrierStore` has no observer machinery yet — apply the HandlerSlot - pattern, then mirror the job wiring; `ReportCarrier` stays gRPC - UNIMPLEMENTED (the stub default) until then. + CANNOT_DO_NOW on unknown job / illegal transition). ✅ Carriers (E87): + `CarrierStore` gained the HandlerSlot observer pattern; the daemon's + id-observer forwards host S3F17 decisions as `CarrierAction` (PROCEED on + Confirmed, CANCEL on CancelCarrier) and `ReportCarrier` drives the + arrival/access flow (WAITING creates + slot map; IN_ACCESS/COMPLETE the + access FSM). KNOWN EDGE: a host CancelCarrier on a still-NotConfirmed + carrier is an FSM self-transition, so no observer fires and the tool + isn't notified — fix needs an event-level (not state-change) hook; + low priority (hosts cancel after a read, i.e. from Confirmed). 11. ✅ `ProcessProgram` on the stream when S7F3 lands (new RecipeStore added-observer) and `ConstantChange` when S2F15 is ACCEPTED (new EquipmentConstantStore changed-observer; rejected writes never fire). @@ -242,9 +247,21 @@ debts tax every later phase, and the most valuable tests aren't automated. DynamicUser/ProtectSystem/StateDirectory/TimeoutStopSec). All enforced by `tools/check_daemon_ops.sh`. Single-session (HSMS-SS) assumption documented in ch42 §5. -16. ⬜ Remaining Layer-1 API: traces, limits, substrates/modules, terminal - services, spool flush RPC, `Describe` RPC. (Engine-side all exists; - surface on demand.) +16. 🚧 Surfaced: `Describe` (full name inventory), `FlushSpool` (purge/ + drain), `SendTerminalMessage` (S10F1 tool→host). ⬜ Still engine-only: + traces (S2F23), limits (S2F45), substrate/E90 + module/E157 tracking — + surface on demand. +17. ✅ Stream responsiveness: Subscribe/WatchHealth poll at 100ms (was + 500ms) so a cancelled stream frees its sync-server worker thread + promptly — found via the randomized stress test below. +18. ✅ Test tiers added: a randomized concurrent RPC stress case (4 threads + × 250 seeded ops incl. Subscribe churn; TSan target; prints its seed) + and a **virtual fab** (`interop/virtual_fab.py` + the `fab` compose + service / `tools/spawn_fab.sh`): N daemons, each with a secsgem-py host + AND a secsgem_client tool, driven by seeded random traffic with + end-to-end invariant checks (round-trips, S6F11/S5F1 delivery, the + command loop). Wired into `run_interop.sh` (now 13 steps). Verified + green at N=3. ### Phase F — fab acceptance (parallel track; the hard gate) - ⚠️ **Standards correctness remains unverified against SEMI texts** (behaviour diff --git a/include/secsgem/daemon/equipment_service.hpp b/include/secsgem/daemon/equipment_service.hpp index c2f4990..b847ed1 100644 --- a/include/secsgem/daemon/equipment_service.hpp +++ b/include/secsgem/daemon/equipment_service.hpp @@ -26,6 +26,7 @@ #include #include +#include "secsgem/gem/messages.hpp" #include "secsgem/gem/runtime.hpp" #include "secsgem/secs2/item.hpp" #include "secsgem/v1/equipment.grpc.pb.h" @@ -227,6 +228,33 @@ class EquipmentService final : public pb::Equipment::Service { hr.mutable_process_program()->set_body(body); push_request(std::move(hr)); }); + for (const auto& rcmd : rt.model().commands.names()) commands_.push_back(rcmd); + descr_model_ = rt.descriptor().model_name; + descr_rev_ = rt.descriptor().software_rev; + + // E87 (D10): the host's carrier-ID decisions go to the tool. PROCEED on + // a Confirmed transition (ProceedWithCarrier or Bind), CANCEL on + // CancelCarrier. The tool announces carriers + drives access via + // ReportCarrier. + rt.model().carriers.add_id_handler( + [this](const std::string& cid, gem::CarrierIDStatus, + gem::CarrierIDStatus to, gem::CarrierIDEvent ev) { + pb::CarrierAction::Action action; + if (ev == gem::CarrierIDEvent::CancelCarrier) + action = pb::CarrierAction::CANCEL; + else if (to == gem::CarrierIDStatus::Confirmed) + action = pb::CarrierAction::PROCEED; + else + return; + pb::HostRequest hr; + auto* ca = hr.mutable_carrier(); + ca->set_carrier_id(cid); + if (const auto* c = rt_.model().carriers.get(cid)) + ca->set_port(c->port_id); + ca->set_action(action); + push_request(std::move(hr)); + }); + rt.model().ecids.add_changed_handler( [this](uint32_t ecid, const s2::Item& value) { pb::HostRequest hr; @@ -374,11 +402,14 @@ class EquipmentService final : public pb::Equipment::Service { rt_.log("tool client subscribed" + (req->client().empty() ? "" : " (" + req->client() + ")")); bool alive = true; + // Poll at 100ms: with the sync server one thread is parked here per open + // stream, so a cancelled client must free its worker promptly (gRPC's + // sync API only exposes cancellation via polled IsCancelled()). while (alive && !ctx->IsCancelled()) { std::optional next; { std::unique_lock lk(subs_mu_); - subs_cv_.wait_for(lk, std::chrono::milliseconds(500), + subs_cv_.wait_for(lk, std::chrono::milliseconds(100), [&] { return !sub->queue.empty(); }); if (!sub->queue.empty()) { next = std::move(sub->queue.front()); @@ -452,6 +483,95 @@ class EquipmentService final : public pb::Equipment::Service { return grpc::Status::OK; } + // E87 carrier reporting (observe-and-report): WAITING announces an + // arrived carrier (idempotent; updates the slot map), IN_ACCESS/COMPLETE + // drive the access FSM. See the proto for the contract. + grpc::Status ReportCarrier(grpc::ServerContext*, const pb::CarrierState* req, + pb::Ack* resp) override { + const std::string cid = req->carrier_id(); + const auto port = static_cast(req->port()); + const auto state = req->state(); + std::vector slots(req->slots().begin(), req->slots().end()); + auto outcome = rt_.read_sync( + [this, cid, port, state, slots]() -> std::optional { + auto& carriers = rt_.model().carriers; + if (state == pb::CarrierState::WAITING) { + carriers.create(cid, port, + slots.empty() ? 25 : slots.size()); // idempotent + // E87 slot-map bytes: 0 = Empty, 1 = NotEmpty. + for (std::size_t i = 0; i < slots.size(); ++i) + carriers.set_slot_state(cid, i, slots[i] ? 1 : 0); + if (!slots.empty()) + carriers.fire_slot_map_event(cid, gem::SlotMapEvent::Read); + return true; + } + if (!carriers.has(cid)) return std::nullopt; + if (state == pb::CarrierState::IN_ACCESS) + return carriers.fire_access_event(cid, gem::CarrierAccessEvent::BeginAccess); + if (state == pb::CarrierState::COMPLETE) + return carriers.fire_access_event(cid, gem::CarrierAccessEvent::EndAccess); + return true; + }); + if (!outcome) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("engine io thread did not answer (not running?)"); + } else if (!*outcome) { + resp->set_code(pb::Ack::INVALID_OBJECT); + resp->set_message("no carrier '" + cid + "' (announce it with WAITING first)"); + } else if (!**outcome) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("E87 access table rejects that transition"); + } else { + resp->set_code(pb::Ack::ACCEPT); + } + return grpc::Status::OK; + } + + grpc::Status Describe(grpc::ServerContext*, const pb::Empty*, + pb::EquipmentDescription* resp) override { + resp->set_model_name(descr_model_); + resp->set_software_rev(descr_rev_); + for (const auto& [name, _] : vars_) resp->add_variables(name); + for (const auto& [name, _] : events_) resp->add_events(name); + for (const auto& [name, _] : alarms_) resp->add_alarms(name); + for (const auto& rcmd : commands_) resp->add_commands(rcmd); + for (const auto& [_, name] : ecnames_) resp->add_constants(name); + return grpc::Status::OK; + } + + grpc::Status FlushSpool(grpc::ServerContext*, const pb::SpoolFlushRequest* req, + pb::Ack* resp) override { + const bool purge = req->purge(); + auto done = rt_.read_sync([this, purge]() { + if (purge) rt_.model().spool.clear(); + else rt_.model().spool.drain(); // toward the host (see S6F23) + return true; + }); + resp->set_code(done ? pb::Ack::ACCEPT : pb::Ack::CANNOT_DO_NOW); + if (!done) resp->set_message("engine io thread did not answer"); + return grpc::Status::OK; + } + + grpc::Status SendTerminalMessage(grpc::ServerContext*, const pb::TerminalMessage* req, + pb::Ack* resp) override { + const auto tid = static_cast(req->tid()); + const std::string text = req->text(); + auto delivered = rt_.read_sync([this, tid, text]() { + return rt_.deliver_or_spool( + gem::s10f1_terminal_display_single(tid, text), "S10F1 (tool)"); + }); + if (!delivered) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("engine io thread did not answer"); + } else if (!*delivered) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("no host connected and stream 10 is not spoolable"); + } else { + resp->set_code(pb::Ack::ACCEPT); + } + return grpc::Status::OK; + } + // Operator-panel control-state transitions (e.g. "offline for maintenance"). // Fires the operator events on the io thread and reports what the E30 table // actually did: ACCEPT iff the equipment landed in the requested state, @@ -609,6 +729,8 @@ class EquipmentService final : public pb::Equipment::Service { std::map events_; // CEID by name std::map alarms_; // ALID by name AND stringified id std::map ecnames_; // ECID -> name (ConstantChange) + std::vector commands_; // RCMDs (Describe) + std::string descr_model_, descr_rev_; // device header (Describe) // WatchHealth plumbing: observers (io thread) bump the version and wake the // per-stream wait loops (gRPC threads). diff --git a/include/secsgem/gem/store/carriers.hpp b/include/secsgem/gem/store/carriers.hpp index 375ad72..3b97b75 100644 --- a/include/secsgem/gem/store/carriers.hpp +++ b/include/secsgem/gem/store/carriers.hpp @@ -14,6 +14,7 @@ #include #include +#include "secsgem/gem/handler_slot.hpp" #include "secsgem/gem/carrier_state.hpp" #include "secsgem/gem/load_port_state.hpp" @@ -62,9 +63,15 @@ class CarrierStore { CarrierStore(CarrierStore&&) = delete; CarrierStore& operator=(CarrierStore&&) = delete; + // set_ replaces the primary handler (legacy semantics); add_ appends an + // observer that survives set_ calls (see handler_slot.hpp). The gRPC + // daemon observes via add_ to forward E87 actions onto its tool stream. void set_id_handler(IDChangeHandler h) { on_id_ = std::move(h); } void set_slot_map_handler(SlotMapChangeHandler h) { on_sm_ = std::move(h); } void set_access_handler(AccessChangeHandler h) { on_acc_ = std::move(h); } + void add_id_handler(IDChangeHandler h) { on_id_.add(std::move(h)); } + void add_slot_map_handler(SlotMapChangeHandler h) { on_sm_.add(std::move(h)); } + void add_access_handler(AccessChangeHandler h) { on_acc_.add(std::move(h)); } enum class CreateResult { Created, Denied_AlreadyExists }; @@ -301,9 +308,12 @@ class CarrierStore { } std::map carriers_; - IDChangeHandler on_id_; - SlotMapChangeHandler on_sm_; - AccessChangeHandler on_acc_; + HandlerSlot on_id_; + HandlerSlot on_sm_; + HandlerSlot on_acc_; bool persistent_ = false; std::filesystem::path journal_dir_; uint64_t next_seq_ = 0; diff --git a/interop/virtual_fab.py b/interop/virtual_fab.py new file mode 100644 index 0000000..3190867 --- /dev/null +++ b/interop/virtual_fab.py @@ -0,0 +1,209 @@ +"""A randomized virtual fab. + +N secs_gemd equipment instances (the `fab` compose service) each get TWO +independent actors attached: + + - a secsgem-py GemHostHandler playing the fab host over HSMS, and + - a secsgem_client Equipment playing the tool software over gRPC, + +then a seeded random scenario drives all of them concurrently: variable +writes with read-back verification, event fires that must arrive at the +host as S6F11, alarm set/clear that must arrive as S5F1, host S1F3 polls, +and host S2F41 commands that must come back HCACK=4 and reach the tool's +handler. Every violated invariant is recorded; the seed is printed so any +failure is reproducible. + + python3 virtual_fab.py --host fab --n 3 --seconds 20 [--seed 1234] + +Exit 0 = every invariant held on every equipment. +""" + +from __future__ import annotations + +import argparse +import logging +import random +import sys +import threading +import time + +sys.path.insert(0, "/app/clients/python") + +import secsgem.common +import secsgem.gem +import secsgem.hsms +import secsgem.secs + +from secsgem_client import Equipment, SecsGemError + +LOG = logging.getLogger("fab") +F = secsgem.secs.functions + + +class EquipmentActor: + """One equipment: its host, its tool, its tallies, its verdicts.""" + + def __init__(self, idx: int, host: str, hsms_port: int, grpc_port: int, + rng: random.Random): + self.idx = idx + self.rng = rng + self.violations: list[str] = [] + self.s6f11_seen = 0 + self.s5f1_seen = 0 + self.events_fired = 0 + self.alarm_ops = 0 + self.commands_sent = 0 + self.commands_received = 0 + self.ops = 0 + + self.tool = Equipment(f"{host}:{grpc_port}") + + @self.tool.on("*") + def _any(cmd): # noqa: ANN001 + self.commands_received += 1 + self.tool.fire("ProcessStarted") # completion signal + + self.tool.listen(background=True) + + settings = secsgem.hsms.HsmsSettings( + address=host, port=hsms_port, session_id=0, + connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, + device_type=secsgem.common.DeviceType.HOST) + self.host = secsgem.gem.GemHostHandler(settings) + self.host.register_stream_function(6, 11, self._on_s6f11) + self.host.register_stream_function(5, 1, self._on_s5f1) + self.host.enable() + + def _on_s6f11(self, _h, message): + self.s6f11_seen += 1 + self.host.send_response(F.SecsS06F12(0), message.header.system) + + def _on_s5f1(self, _h, message): + self.s5f1_seen += 1 + self.host.send_response(F.SecsS05F02(0), message.header.system) + + def violate(self, what: str) -> None: + self.violations.append(what) + LOG.error("[eq%d] VIOLATION: %s", self.idx, what) + + def setup(self) -> bool: + if not self.host.waitfor_communicating(timeout=20): + self.violate("HSMS establish-communications failed") + return False + self.host.send_and_waitfor_response(F.SecsS01F17()) + self.host.send_and_waitfor_response( + F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [101]}]})) + self.host.send_and_waitfor_response( + F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]})) + self.host.send_and_waitfor_response( + F.SecsS02F37({"CEED": True, "CEID": [300]})) + self.host.send_and_waitfor_response( + F.SecsS05F03({"ALED": 0x80, "ALID": 1})) + return True + + def step(self) -> None: + """One random action with its invariant.""" + self.ops += 1 + roll = self.rng.random() + try: + if roll < 0.30: + v = round(self.rng.uniform(0.1, 9.9), 3) + self.tool.set(ChamberPressure=v) + got = self.tool.get("ChamberPressure")["ChamberPressure"] + if abs(got - v) > 1e-3: + self.violate(f"round-trip: set {v} got {got}") + elif roll < 0.45: + self.tool.fire("ProcessStarted") + self.events_fired += 1 + elif roll < 0.60: + if self.rng.random() < 0.5: + self.tool.alarm("chiller_temp_high") + else: + self.tool.clear("chiller_temp_high") + self.alarm_ops += 1 + elif roll < 0.80: + rsp = self.host.send_and_waitfor_response(F.SecsS01F03([])) + body = self.host.settings.streams_functions.decode(rsp).get() + if not isinstance(body, list) or len(body) < 1: + self.violate(f"S1F3 poll returned {body!r}") + else: + rsp = self.host.send_and_waitfor_response( + F.SecsS02F41({"RCMD": "START", "PARAMS": []})) + body = self.host.settings.streams_functions.decode(rsp).get() + hcack = int(body.get("HCACK", -1)) if isinstance(body, dict) else -1 + self.commands_sent += 1 + # A tool is subscribed for the whole run: must be HCACK 4. + if hcack != 4: + self.violate(f"S2F41 with subscriber -> HCACK {hcack} (want 4)") + except SecsGemError as e: + self.violate(f"client raised: {e}") + except Exception as e: # noqa: BLE001 + self.violate(f"actor raised: {type(e).__name__}: {e}") + + def finish(self) -> None: + time.sleep(2.0) # let in-flight S6F11/S5F1 land + if self.events_fired + self.commands_received > 0 and self.s6f11_seen == 0: + self.violate("events fired but host never received S6F11") + if self.alarm_ops > 0 and self.s5f1_seen == 0: + self.violate("alarms toggled but host never received S5F1") + if self.commands_sent > 0 and self.commands_received == 0: + self.violate("commands sent but tool handler never ran") + try: + self.host.disable() + self.tool.close() + except Exception: # noqa: BLE001 + pass + + +def run(host: str, n: int, seconds: float, seed: int) -> int: + LOG.info("virtual fab: %d equipment, %.0fs, seed=%d", n, seconds, seed) + actors = [EquipmentActor(i, host, 5100 + i, 51000 + i, + random.Random(seed + i)) for i in range(n)] + if not all(a.setup() for a in actors): + return 1 + + deadline = time.monotonic() + seconds + + def drive(a: EquipmentActor) -> None: + while time.monotonic() < deadline and len(a.violations) < 5: + a.step() + time.sleep(a.rng.uniform(0.01, 0.10)) + + threads = [threading.Thread(target=drive, args=(a,)) for a in actors] + for t in threads: + t.start() + for t in threads: + t.join() + for a in actors: + a.finish() + + bad = 0 + for a in actors: + LOG.info("[eq%d] ops=%d cmds=%d/%d s6f11=%d s5f1=%d violations=%d", + a.idx, a.ops, a.commands_received, a.commands_sent, + a.s6f11_seen, a.s5f1_seen, len(a.violations)) + bad += len(a.violations) + if bad: + LOG.error("FAB FAILED: %d violations (seed=%d to reproduce)", bad, seed) + return 1 + LOG.info("virtual fab: all invariants held on all %d equipment (seed=%d)", + n, seed) + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--host", default="fab") + ap.add_argument("--n", type=int, default=3) + ap.add_argument("--seconds", type=float, default=20.0) + ap.add_argument("--seed", type=int, default=None) + args = ap.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s") + for noisy in ("communication", "hsms_connection", "secsgem"): + logging.getLogger(noisy).setLevel(logging.WARNING) + seed = args.seed if args.seed is not None else random.SystemRandom().randrange(1 << 31) + return run(args.host, args.n, args.seconds, seed) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/proto/secsgem/v1/equipment.proto b/proto/secsgem/v1/equipment.proto index d1256bc..dc36529 100644 --- a/proto/secsgem/v1/equipment.proto +++ b/proto/secsgem/v1/equipment.proto @@ -83,13 +83,31 @@ service Equipment { // are long-lived objects you report against as the physical work proceeds. rpc ReportProcessJob(ProcessJobState) returns (Ack); // E40 — job-based tools - rpc ReportCarrier(CarrierState) returns (Ack); // E87 — carrier-based tools + + // E87 — carrier-based tools. WAITING announces a physically-arrived + // carrier (creates it; idempotent — re-announce updates the slot map); + // IN_ACCESS / COMPLETE drive the access FSM. The host's S3F17 decisions + // (ProceedWithCarrier / CancelCarrier) come back on the Subscribe stream + // as CarrierAction. + rpc ReportCarrier(CarrierState) returns (Ack); // ---- Diagnostics -------------------------------------------------------- // Live daemon/link status: distinguishes "host went offline" from "cable // unplugged" from "spool filling up". Streams a snapshot on every change. rpc WatchHealth(Empty) returns (stream Health); + + // Everything this equipment is configured with, by name — for tooling, + // diagnostics, and client-side validation/autocomplete. + rpc Describe(Empty) returns (EquipmentDescription); + + // Flush the spool: purge=true discards queued messages, purge=false drains + // them toward the host (requires a SELECTED session). + rpc FlushSpool(SpoolFlushRequest) returns (Ack); + + // Equipment-initiated operator message to the host (S10F1). Fails with + // CANNOT_DO_NOW when no host is connected and stream 10 isn't spoolable. + rpc SendTerminalMessage(TerminalMessage) returns (Ack); } // ---- Values ---------------------------------------------------------------- @@ -260,6 +278,27 @@ message Health { } } +// ---- Diagnostics & operations ----------------------------------------------- + +message EquipmentDescription { + string model_name = 1; + string software_rev = 2; + repeated string variables = 3; // SVID + DVID names + repeated string events = 4; // collection-event names + repeated string alarms = 5; // alarm names (or stringified ALIDs) + repeated string commands = 6; // RCMDs the host may send + repeated string constants = 7; // equipment-constant names +} + +message SpoolFlushRequest { + bool purge = 1; // true = discard, false = drain to host +} + +message TerminalMessage { + uint32 tid = 1; // terminal id (0 = main) + string text = 2; +} + // ---- Acknowledgement ------------------------------------------------------- // Mirrors SEMI HCACK exactly. For non-command RPCs, only ACCEPT vs an error diff --git a/tests/test_daemon_service.cpp b/tests/test_daemon_service.cpp index e137c31..572c3d6 100644 --- a/tests/test_daemon_service.cpp +++ b/tests/test_daemon_service.cpp @@ -3,8 +3,12 @@ #include +#include #include +#include +#include #include +#include #include "secsgem/daemon/equipment_service.hpp" #include "secsgem/gem/default_handlers.hpp" @@ -487,3 +491,247 @@ TEST_CASE("Phase D: jobs, recipes, and EC changes on the stream; ReportProcessJo server->Shutdown(); rt.stop(); } + +// D10 + E16: carriers on the stream, Describe / FlushSpool / terminal. +TEST_CASE("carriers (D10) and the operations RPCs (E16)") { + gem::EquipmentRuntime rt(test_config()); + gem::register_default_handlers(rt); + dmn::EquipmentService svc(rt); + rt.run_async(); + + grpc::ServerBuilder builder; + builder.RegisterService(&svc); + std::unique_ptr server(builder.BuildAndStart()); + REQUIRE(server); + auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{})); + + grpc::ClientContext sub_ctx; + pb::SubscribeRequest sreq; + auto reader = stub->Subscribe(&sub_ctx, sreq); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + auto report_carrier = [&](const std::string& cid, pb::CarrierState::State st, + std::vector slots = {}) { + grpc::ClientContext ctx; + pb::CarrierState req; + pb::Ack ack; + req.set_carrier_id(cid); + req.set_port(2); + req.set_state(st); + for (bool b : slots) req.add_slots(b); + REQUIRE(stub->ReportCarrier(&ctx, req, &ack).ok()); + return ack.code(); + }; + + // Tool announces an arrived FOUP with a slot map. + CHECK(report_carrier("CAR-D-1", pb::CarrierState::WAITING, + {true, true, false}) == pb::Ack::ACCEPT); + auto created = rt.read_sync([&rt]() { + const auto* c = rt.model().carriers.get("CAR-D-1"); + return c && c->slots.size() == 3 && c->slots[0].state == 1 && + c->slots[2].state == 0 && c->port_id == 2; + }); + REQUIRE(created.has_value()); + CHECK(*created); + + // Host says ProceedWithCarrier (S3F17) -> tool stream gets PROCEED. + auto reply = rt.read_sync([&rt]() { + return rt.router().dispatch( + gem::s3f17_carrier_action(0u, "ProceedWithCarrier", "CAR-D-1", {})); + }); + REQUIRE(reply.has_value()); + REQUIRE(reply->has_value()); + pb::HostRequest hr; + REQUIRE(reader->Read(&hr)); + REQUIRE(hr.has_carrier()); + CHECK(hr.carrier().carrier_id() == "CAR-D-1"); + CHECK(hr.carrier().port() == 2); + CHECK(hr.carrier().action() == pb::CarrierAction::PROCEED); + + // Tool drives access: begin + end; reporting against an unknown id fails. + CHECK(report_carrier("CAR-D-1", pb::CarrierState::IN_ACCESS) == pb::Ack::ACCEPT); + CHECK(report_carrier("CAR-D-1", pb::CarrierState::COMPLETE) == pb::Ack::ACCEPT); + CHECK(report_carrier("CAR-GHOST", pb::CarrierState::IN_ACCESS) == + pb::Ack::INVALID_OBJECT); + + // Host cancels CAR-D-1 (Confirmed -> NotConfirmed is a real transition, so + // the observer fires) -> CANCEL on the stream. NOTE: a CancelCarrier on a + // still-NotConfirmed carrier is a self-transition the FSM doesn't signal, + // so the tool isn't told — a known E87 edge (see DAEMON_ROADMAP). + (void)rt.read_sync([&rt]() { + return rt.router().dispatch( + gem::s3f17_carrier_action(0u, "CancelCarrier", "CAR-D-1", {})); + }); + REQUIRE(reader->Read(&hr)); + REQUIRE(hr.has_carrier()); + CHECK(hr.carrier().action() == pb::CarrierAction::CANCEL); + + // ---- E16: Describe / FlushSpool / SendTerminalMessage -------------------- + { + grpc::ClientContext ctx; + pb::Empty req; + pb::EquipmentDescription d; + REQUIRE(stub->Describe(&ctx, req, &d).ok()); + CHECK(d.model_name() == "SECSGEM-SIM"); + auto has = [](const auto& list, const std::string& want) { + for (const auto& e : list) + if (e == want) return true; + return false; + }; + CHECK(has(d.variables(), "ChamberPressure")); + CHECK(has(d.events(), "ProcessStarted")); + CHECK(has(d.alarms(), "chiller_temp_high")); + CHECK(has(d.commands(), "START")); + CHECK(has(d.constants(), "TimeFormat")); + } + { + // No host: a fired event spools (stream 6 is spoolable); purge empties it. + grpc::ClientContext fctx; + pb::Event ev; + pb::Ack ack; + ev.set_name("ProcessStarted"); + // Enable the event first so emit isn't suppressed. + (void)rt.read_sync([&rt]() { + return rt.model().enable_events(true, {300}); + }); + REQUIRE(stub->FireEvent(&fctx, ev, &ack).ok()); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + auto depth = rt.read_sync([&rt]() { return rt.model().spool.size(); }); + REQUIRE(depth.has_value()); + CHECK(*depth == 1); + + grpc::ClientContext pctx; + pb::SpoolFlushRequest freq; + pb::Ack fack; + freq.set_purge(true); + REQUIRE(stub->FlushSpool(&pctx, freq, &fack).ok()); + CHECK(fack.code() == pb::Ack::ACCEPT); + depth = rt.read_sync([&rt]() { return rt.model().spool.size(); }); + CHECK(*depth == 0); + } + { + // Stream 10 is not spoolable and no host is connected -> honest refusal. + grpc::ClientContext ctx; + pb::TerminalMessage req; + pb::Ack ack; + req.set_tid(0); + req.set_text("hello fab"); + REQUIRE(stub->SendTerminalMessage(&ctx, req, &ack).ok()); + CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW); + } + + sub_ctx.TryCancel(); + pb::HostRequest drain; + while (reader->Read(&drain)) {} + (void)reader->Finish(); + server->Shutdown(); + rt.stop(); +} + +// Randomized concurrent stress: several client threads fire a seeded random +// mix of RPCs against the live service while the io thread runs — the +// strongest TSan target we have, and a probe for ordering/lifetime bugs the +// scenario tests can't reach. Failures print the seed for reproduction. +TEST_CASE("randomized concurrent RPC stress (seeded)") { + const unsigned seed = static_cast( + std::random_device{}()); // logged below; rerun by hardcoding it + INFO("stress seed = " << seed); + + gem::EquipmentRuntime rt(test_config()); + gem::register_default_handlers(rt); + dmn::EquipmentService svc(rt); + rt.run_async(); + + grpc::ServerBuilder builder; + builder.RegisterService(&svc); + std::unique_ptr server(builder.BuildAndStart()); + REQUIRE(server); + auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{})); + + constexpr int kThreads = 4; + constexpr int kOpsPerThread = 250; + std::atomic failures{0}; + + auto worker = [&](unsigned tseed) { + std::mt19937 rng(tseed); + std::uniform_int_distribution op(0, 7); + std::uniform_real_distribution val(0.0, 10.0); + for (int i = 0; i < kOpsPerThread; ++i) { + grpc::ClientContext ctx; + pb::Ack ack; + grpc::Status st = grpc::Status::OK; + switch (op(rng)) { + case 0: { + pb::VariableUpdate r; + (*r.mutable_values())["ChamberPressure"].set_real(val(rng)); + st = stub->SetVariables(&ctx, r, &ack); + break; + } + case 1: { + pb::VariableQuery r; + pb::VariableSnapshot snap; + st = stub->GetVariables(&ctx, r, &snap); + if (st.ok() && snap.values().empty()) ++failures; // config never empty + break; + } + case 2: { + pb::Event r; + r.set_name("ProcessStarted"); + st = stub->FireEvent(&ctx, r, &ack); + break; + } + case 3: { + pb::Alarm r; + r.set_name("chiller_temp_high"); + st = stub->SetAlarm(&ctx, r, &ack); + break; + } + case 4: { + pb::Alarm r; + r.set_name("chiller_temp_high"); + st = stub->ClearAlarm(&ctx, r, &ack); + break; + } + case 5: { + pb::Empty r; + pb::ControlState cs; + st = stub->GetControlState(&ctx, r, &cs); + break; + } + case 6: { + pb::Empty r; + pb::EquipmentDescription d; + st = stub->Describe(&ctx, r, &d); + if (st.ok() && d.variables_size() == 0) ++failures; + break; + } + case 7: { + // Subscribe/cancel churn: exercises subscriber add/remove under load. + grpc::ClientContext sctx; + pb::SubscribeRequest sr; + auto rd = stub->Subscribe(&sctx, sr); + sctx.TryCancel(); + pb::HostRequest hr; + while (rd->Read(&hr)) {} + (void)rd->Finish(); + break; + } + } + if (!st.ok()) ++failures; + } + }; + + std::vector threads; + for (int t = 0; t < kThreads; ++t) threads.emplace_back(worker, seed + t); + for (auto& th : threads) th.join(); + + CHECK(failures.load() == 0); + // The engine must still be fully responsive afterwards. + grpc::ClientContext ctx; + pb::Empty req; + pb::ControlState cs; + CHECK(stub->GetControlState(&ctx, req, &cs).ok()); + + server->Shutdown(); + rt.stop(); +} diff --git a/tools/run_interop.sh b/tools/run_interop.sh index 9397ce4..d27c744 100755 --- a/tools/run_interop.sh +++ b/tools/run_interop.sh @@ -19,6 +19,7 @@ # pyclient published secsgem-client package vs secs_gemd (13 checks) # spool spool persistence across a server restart # daemon-ops unix-socket gRPC + Prometheus gauges + graceful SIGTERM +# fab randomized virtual fab: FAB_N daemons x (host + tool) actors # tshark Wireshark HSMS dissector on a live capture # secs4j secs4java8 Java host vs C++ secs_server (55 checks) # @@ -37,7 +38,7 @@ record() { # record compose() { docker compose "$@"; } -stop_services() { compose stop server gemd server-spool >/dev/null 2>&1 || true; } +stop_services() { compose stop server gemd server-spool fab >/dev/null 2>&1 || true; } trap stop_services EXIT # ---- build ----------------------------------------------------------------- @@ -126,6 +127,15 @@ note "daemon-ops: unix socket, Prometheus gauges, SIGTERM drains cleanly" compose run --rm builder bash tools/check_daemon_ops.sh record daemon-ops $? +# ---- randomized virtual fab ---------------------------------------------------- +note "fab: ${FAB_N:-3} equipment x (secsgem-py host + secsgem-client tool), seeded random traffic" +compose up -d --no-deps fab +sleep 3 +compose run --rm --no-deps -e PYTHONPATH=/app/clients/python interop \ + python3 virtual_fab.py --host fab --n "${FAB_N:-3}" --seconds "${FAB_SECONDS:-20}" +record fab $? +compose stop fab >/dev/null 2>&1 + # ---- Wireshark dissector ------------------------------------------------------ note "tshark: Wireshark HSMS dissector" compose run --rm builder bash interop/tshark_validate.sh diff --git a/tools/spawn_fab.sh b/tools/spawn_fab.sh new file mode 100755 index 0000000..836ebd6 --- /dev/null +++ b/tools/spawn_fab.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Spawn a virtual fab: N secs_gemd equipment instances in one container. +# equipment i: HSMS on 5100+i, gRPC on 0.0.0.0:51000+i (docker-network +# internal — the localhost default is for real deployments). +# Used by the `fab` compose service; interop/virtual_fab.py drives them. +set -u +cd "$(dirname "${BASH_SOURCE[0]}")/.." +N=${FAB_N:-3} +PIDS=() +trap 'kill "${PIDS[@]}" 2>/dev/null; wait' TERM INT +for i in $(seq 0 $((N - 1))); do + ./build/secs_gemd \ + --port $((5100 + i)) \ + --grpc "0.0.0.0:$((51000 + i))" \ + --config-dir data \ + > "/tmp/gemd-$i.log" 2>&1 & + PIDS+=($!) +done +echo "virtual fab up: $N equipment (HSMS 5100+, gRPC 51000+)" +wait