raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: live persistent-spool restart end-to-end

Browse Source 
Adds a docker-compose service `server-spool` that runs secs_server
with --spool-dir pointed at a named volume.  Two-phase Python
harness (interop/spool_persistence_test.py):

  1. Enqueue phase: force-spool one S6F11(CEID=300) via the
     SPOOL_ON / START / SPOOL_OFF RCMD trio, then disconnect.
  2. Driver runs `docker compose restart server-spool` between
     the phases — the named volume preserves the journal files.
  3. Drain phase: reconnect, send S6F23(Transmit), verify the
     replayed S6F11 carries CEID 300.

Surfaces a real interop bug along the way: secsgem-py 0.3.0 encodes
RSDC (and other "single-byte status" fields) as <U1>, while SEMI E5
spells them as <B>.  Our `as_binary_first` was strict on Binary; now
accepts either (the byte semantics are identical, and the leniency is
symmetric with the U-type widening from the first interop commit).

Result: enqueue → docker restart → drain returns CEID 300 cleanly.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-09 01:04:49 +02:00
parent df6060a9e9
commit a1dc7937d4
3 changed files with 263 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker-compose.yml
+23
View File
@@ -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: {}
include/secsgem/gem/messages_helpers.hpp
+13 -1
View File
@@ -51,12 +51,24 @@ inline std::optional<std::string> as_text_or_binary(const s2::Item& item) {
return std::nullopt;
}
// Single-byte status/ack reader. SEMI E5 spells these fields as
// `<B>` (Binary), but real peers — notably secsgem-py 0.3.0 — encode
// RSDC, EAC, ACKC*, OFLACK, ONLACK, etc. as `<U1>`. 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<uint8_t> as_binary_first(const s2::Item& item) {
if (item.format() != s2::Format::Binary) return std::nullopt;
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<std::vector<uint8_t>>(item.storage());
if (v.empty()) return std::nullopt;
return v.front();
}
return std::nullopt;
}
inline std::optional<bool> as_boolean(const s2::Item& item) {
if (item.format() != s2::Format::Boolean) return std::nullopt;
interop/spool_persistence_test.py
+224
View File
@@ -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())