diff --git a/docker-compose.yml b/docker-compose.yml index dff4a9f..e289577 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -75,8 +75,31 @@ services: command: ["python3", "/app/interop/passive_equipment.py", "--port", "5000"] networks: [secs] + # Server variant for the persistent-spool restart test. Identical to + # `server` but with --spool-dir pointed at a named volume so spooled + # messages survive `docker compose restart server-spool`. + server-spool: + <<: *base + depends_on: + builder: + condition: service_completed_successfully + command: + - /app/build/secs_server + - --port + - "5000" + - --device + - "0" + - --spool-dir + - /spool + volumes: + - .:/app + - build:/app/build + - spool:/spool + networks: [secs] + networks: secs: {} volumes: build: {} + spool: {} diff --git a/include/secsgem/gem/messages_helpers.hpp b/include/secsgem/gem/messages_helpers.hpp index 6901287..a9ef1cf 100644 --- a/include/secsgem/gem/messages_helpers.hpp +++ b/include/secsgem/gem/messages_helpers.hpp @@ -51,11 +51,23 @@ inline std::optional as_text_or_binary(const s2::Item& item) { return std::nullopt; } +// Single-byte status/ack reader. SEMI E5 spells these fields as +// `` (Binary), but real peers — notably secsgem-py 0.3.0 — encode +// RSDC, EAC, ACKC*, OFLACK, ONLACK, etc. as ``. Both formats +// carry the same value in the same single byte, so we accept either +// and return `nullopt` only when the item is genuinely neither. inline std::optional as_binary_first(const s2::Item& item) { - if (item.format() != s2::Format::Binary) return std::nullopt; - const auto& v = item.as_bytes(); - if (v.empty()) return std::nullopt; - return v.front(); + if (item.format() == s2::Format::Binary) { + const auto& v = item.as_bytes(); + if (v.empty()) return std::nullopt; + return v.front(); + } + if (item.format() == s2::Format::U1) { + const auto& v = std::get>(item.storage()); + if (v.empty()) return std::nullopt; + return v.front(); + } + return std::nullopt; } inline std::optional as_boolean(const s2::Item& item) { diff --git a/interop/spool_persistence_test.py b/interop/spool_persistence_test.py new file mode 100644 index 0000000..e7e4925 --- /dev/null +++ b/interop/spool_persistence_test.py @@ -0,0 +1,224 @@ +"""End-to-end persistent-spool restart test. + +Exercises SpoolStore's opt-in file-backed journal by: + + 1. Connecting to the C++ server (which has been started with + --spool-dir /spool, a docker named volume). + 2. Force-spooling a CEID 300 emission via the SPOOL_ON / START / + SPOOL_OFF RCMD sequence. + 3. Disconnecting cleanly (separate). + 4. The caller (a shell driver) then issues `docker compose restart + server-spool`, preserving the /spool volume. + 5. We reconnect. If persistence works, the server should auto-emit + S6F25 (spool-data-ready) shortly after SELECT. We then issue + S6F23 RSDC=Transmit and expect the queued S6F11 to drain. + +Two modes: + + python3 spool_persistence_test.py --phase enqueue --host server-spool + python3 spool_persistence_test.py --phase drain --host server-spool + +The driver runs phase 1, restarts the container, runs phase 2. Each +phase returns 0 on success. +""" + +from __future__ import annotations + +import argparse +import logging +import sys +import threading +import time + +import secsgem.common +import secsgem.gem +import secsgem.hsms +import secsgem.secs.functions as F +from secsgem.secs.functions.base import SecsStreamFunction + + +LOG = logging.getLogger("interop.spool-restart") + + +# secsgem-py 0.3.0 ships no S6F25/F26 — we register them inline so the +# protocol layer can decode the spool-data-ready notification. +class S06F25(SecsStreamFunction): + _stream = 6 + _function = 25 + _to_host = True + _to_equipment = False + _has_reply = True + _is_reply_required = True + _is_multi_block = False + _data_format = "< DATAID >" # carries the queued-message count as a U4 + + +class S06F26(SecsStreamFunction): + _stream = 6 + _function = 26 + _to_host = False + _to_equipment = True + _has_reply = False + _is_reply_required = False + _is_multi_block = False + _data_format = "< ACKC6 >" + + + + +def connect(host: str, port: int, session_id: int) -> secsgem.gem.GemHostHandler: + settings = secsgem.hsms.HsmsSettings( + address=host, + port=port, + session_id=session_id, + connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, + device_type=secsgem.common.DeviceType.HOST, + ) + client = secsgem.gem.GemHostHandler(settings) + client.enable() + if not client.waitfor_communicating(timeout=15): + raise RuntimeError("failed to communicate within 15s") + return client + + +def rcmd(client, name: str) -> int: + rsp = client.send_and_waitfor_response( + F.SecsS02F41({"RCMD": name, "PARAMS": []}) + ) + body = client.settings.streams_functions.decode(rsp).get() + return body.get("HCACK") if isinstance(body, dict) else -1 + + +def phase_enqueue(host: str, port: int, session_id: int) -> int: + """Enqueue one spooled S6F11 then disconnect.""" + client = connect(host, port, session_id) + try: + # Define + link + enable CEID 300 so the equipment actually emits S6F11. + client.send_and_waitfor_response( + F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [2]}]}) + ) + client.send_and_waitfor_response( + F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]}) + ) + client.send_and_waitfor_response( + F.SecsS02F37({"CEED": True, "CEID": [300]}) + ) + + # Force the spool on, fire START (emits CEID 300, goes to spool), + # then turn force off. The S6F11 is now persisted on disk. + ack = rcmd(client, "SPOOL_ON") + if ack != 0: + LOG.error("SPOOL_ON ack=%s", ack); return 1 + ack = rcmd(client, "START") + if ack != 0: + LOG.error("START ack=%s", ack); return 1 + ack = rcmd(client, "SPOOL_OFF") + if ack != 0: + LOG.error("SPOOL_OFF ack=%s", ack); return 1 + + LOG.info("phase=enqueue: spooled one S6F11 (CEID=300)") + # Brief pause so the post-RCMD spool write completes before we close. + time.sleep(0.3) + return 0 + finally: + try: client.disable() + except Exception: pass + time.sleep(0.2) + + +def phase_drain(host: str, port: int, session_id: int) -> int: + """After server restart: drain via S6F23 and verify the replayed S6F11. + + Note: the C++ server also auto-emits S6F25 (spool-data-ready) on + re-SELECT, but secsgem-py 0.3.0 doesn't ship S6F25/F26 in its + catalog and decoding ours runs into SFDL grammar trouble; the + persistence claim doesn't actually depend on F25 (it's just a hint + to the host), so we skip the F25 wait and request transmit + directly. If the spool was empty after restart, S6F23(Transmit) + would ack but no S6F11 would arrive — that's the real test. + """ + drained_ceids: list[int] = [] + drained_event = threading.Event() + + settings = secsgem.hsms.HsmsSettings( + address=host, port=port, session_id=session_id, + connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, + device_type=secsgem.common.DeviceType.HOST, + ) + client = secsgem.gem.GemHostHandler(settings) + # Register S6F25 / S6F26 so the protocol layer can at least decode + # the unsolicited notification cleanly while we wait on the drain. + client.settings.streams_functions.update(S06F25) + client.settings.streams_functions.update(S06F26) + + def on_s6f25(_handler, message): + LOG.info("[evt] S6F25 spool-data-ready (server announces queue)") + client.send_response(S06F26(0), message.header.system) + + def on_s6f11(_handler, message): + decoded = client.settings.streams_functions.decode(message) + body = decoded.get() + ceid = body.get("CEID") if isinstance(body, dict) else None + if ceid is not None: + drained_ceids.append(int(ceid)) + LOG.info("[evt] replayed S6F11 CEID=%s", ceid) + drained_event.set() + client.send_response(F.SecsS06F12(0), message.header.system) + + client.register_stream_function(6, 25, on_s6f25) + client.register_stream_function(6, 11, on_s6f11) + + client.enable() + try: + if not client.waitfor_communicating(timeout=15): + LOG.error("waitfor_communicating timed out"); return 1 + + # Brief pause so any auto-emitted S6F25 lands before we drain. + time.sleep(0.5) + + # Ask the server to drain. + rsp = client.send_and_waitfor_response(F.SecsS06F23(0)) # RSDC=Transmit + body = client.settings.streams_functions.decode(rsp).get() + rsda = body if isinstance(body, int) else -1 + LOG.info("S6F23 -> S6F24 RSDA=%s", rsda) + if rsda != 0: + LOG.error("RSDA != Accept"); return 1 + + # Wait for the replayed S6F11 (CEID=300). If persistence broke, + # there'd be no replay here even though the ack said Accept. + if not drained_event.wait(timeout=3.0): + LOG.error("no S6F11 replayed within 3s — spool empty after restart?") + return 1 + if 300 not in drained_ceids: + LOG.error("replayed CEIDs=%s, expected 300", drained_ceids) + return 1 + + LOG.info("phase=drain: spool replayed cleanly — persistence works") + return 0 + finally: + try: client.disable() + except Exception: pass + time.sleep(0.2) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--phase", choices=["enqueue", "drain"], required=True) + ap.add_argument("--host", default="server-spool") + ap.add_argument("--port", type=int, default=5000) + ap.add_argument("--session-id", type=int, default=0) + ap.add_argument("--log-level", default="INFO") + args = ap.parse_args() + logging.basicConfig(level=args.log_level, + format="%(asctime)s %(levelname)s %(name)s: %(message)s") + if args.log_level.upper() != "DEBUG": + logging.getLogger("communication").setLevel(logging.WARNING) + logging.getLogger("hsms_connection").setLevel(logging.WARNING) + + if args.phase == "enqueue": + return phase_enqueue(args.host, args.port, args.session_id) + return phase_drain(args.host, args.port, args.session_id) + + +if __name__ == "__main__": + sys.exit(main())