dispatcher: subscribe before the opening rows, and ADR-0058's retired rule (#374 #367)

docs/adr/0058-hardware-meets-the-bus-and-a-translator-is-only-for-hardware-that-cannot.md

@@ -10,17 +10,24 @@ translator is a layer everything passes through. Neither holds.
 
 ## What ADR-0043 already settles, and this does not touch
 
-An address names its system as its first level — `dccex/12`, `jmri/LT3` — and
-the topic carries the system as a level, so **whatever answers for a system
-subscribes that system and an address nothing answers to does no harm**, as a
-DCC packet nobody picks up does. No ownership table exists anywhere. Zero, one
-or several answer, per what is wired. State rather than command, so something
-coming up finds positions to set rather than a history to replay. Any detector
-meets the same door through a republisher.
+**Whatever answers for an address acts on it, and an address nothing answers
+to does no harm**, as a DCC packet nobody picks up does. No ownership table
+exists anywhere. Zero, one or several answer, per what is wired. State rather
+than command, so something coming up finds positions to set rather than a
+history to replay. Any detector meets the same door through a republisher.
 
 That is already a design in which the hardware is somebody else's business.
-The code agrees: a point's address is checked for shape and never against a
-list of known systems, so `mine/7` derives today.
+The code agrees: an address is any non-empty string, checked against no list
+of known systems, so `mine/7` and `5` both derive today.
+
+*Amended by [#367](https://github.com/rails49/control/issues/367),
+2026-09-04:* this section used to say an address names its system as its first
+level — `dccex/12`, `jmri/LT3` — and that the topic carries the system as a
+level. Both are withdrawn. An address is the string the drawing carries and
+the hardware answers to, the topic carries it as trailing levels with no
+system in front, and a translator acts on every address it recognises. Two
+systems that number a point alike both act, which is the deployer's addressing
+to fix and not a topic level.
 
 ## The bus is the interface, and a translator is not a layer
 

src/tc49/dispatcher/dispatch.py

@@ -556,6 +556,17 @@ def __init__(
             # a fault that looks like a hang (ADR-0037).
             run=HELD if picture or not adopted.standing else RUNNING,
         )
+        # Before the opening rows and not after. Over a broker a publish is
+        # asynchronous where a subscribe waits for the broker to acknowledge,
+        # so publishing first opens a window a round trip wide in which a
+        # gesture addressed to this app is lost — `state/run` is what a client
+        # waits for to know the dispatcher is up, and a gesture is an event
+        # that nothing replays. In one process the window had no width and the
+        # order did not show. Safe here because `_on_dispatch` already ignores
+        # this app's own announcements coming back past it, and nothing on
+        # `tc49/layout/#` is published from this constructor at all.
+        bus.subscribe("tc49/layout/#", self._on_layout)
+        bus.subscribe("tc49/dispatch/#", self._on_dispatch)
         for train, at in adopted.standing.items():
             self._state.locks[at] = train
             self._state.block_of[train] = at
@@ -594,8 +605,6 @@ def __init__(
         self._publish_aspects()
         self._publish_allocation()
         self._publish_disputed()
-        bus.subscribe("tc49/layout/#", self._on_layout)
-        bus.subscribe("tc49/dispatch/#", self._on_dispatch)
 
     # -- live state, for the property tests' oracles ------------------------
 

tests/dispatcher/test_adoption.py

@@ -27,7 +27,7 @@
 from tc49.lib.bus import InProcessBus, Payload
 from tc49.lib.roster import Train
 from tc49.lib.scenario import TrainSpec
-from tests.harness import load, retaining
+from tests.harness import Recording, load, retaining, subscribed_before_publishing
 
 RUN = "tc49/dispatch/state/run"
 REQUESTS = "tc49/dispatch/request_submitted"
@@ -472,3 +472,22 @@ def test_a_train_the_picture_names_unreadably_loses_only_itself(
     assert dispatcher.state.block_of == {"express_2": "up_w", "freight_1": "yard_w"}
     assert dispatcher.state.crossing == {"express_2": "crossover.up_straight"}
     coherent(dispatcher)
+
+
+def test_it_subscribes_before_it_publishes_anything() -> None:
+    """The opening rows go out with the subscriptions already live.
+
+    `state/run` is what a client waits for to learn the dispatcher is up, so a
+    gesture — a request, a hold, a placement — can arrive the instant after
+    it. Over a broker a publish is asynchronous where a subscribe waits to be
+    acknowledged, so publishing first opens a window a round trip wide in
+    which that gesture is dropped, and an event is not retained so nothing
+    replays it. In one process the window has no width, which is why the
+    order went unnoticed until the apps ran against a real broker.
+    """
+    layout, roster, scenario = load("crossover-yard/meet")
+    bus = Recording()
+    Dispatcher(
+        bus, layout, roster, placement(scenario.trains), FullRoute(layout, DEFAULT_K)
+    )
+    subscribed_before_publishing(bus)

tests/harness.py

@@ -8,7 +8,7 @@
 import json
 import shutil
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 import pytest
 
@@ -225,3 +225,38 @@ def run_rows(assembly: Assembly) -> list[tuple[str, bool]]:
     return [
         (str(line["run"]), bool(line.get("moving"))) for line in leaves(assembly, "run")
     ]
+
+
+class Recording(InProcessBus):
+    """An in-process bus that remembers the order it was called in.
+
+    For the one thing a functional test cannot see: an app must subscribe
+    before it publishes its opening rows. Over a broker a publish is
+    asynchronous where a subscribe waits to be acknowledged, so publishing
+    first drops any gesture arriving in the gap, and an event is not retained.
+    In one process the gap has no width, so the order is asserted here rather
+    than waited for.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(Clock())
+        self.calls: list[str] = []
+
+    def subscribe(self, topic_filter: str, handler: object) -> None:  # type: ignore[override]
+        self.calls.append(f"subscribe {topic_filter}")
+        super().subscribe(topic_filter, cast(object, handler))  # type: ignore[arg-type]
+
+    def publish(self, topic: str, payload: Payload) -> None:
+        self.calls.append(f"publish {topic}")
+        super().publish(topic, payload)
+
+
+def subscribed_before_publishing(bus: Recording) -> None:
+    """Assert it of whatever was just built on `bus`."""
+    published = [i for i, call in enumerate(bus.calls) if call.startswith("publish")]
+    subscribed = [i for i, call in enumerate(bus.calls) if call.startswith("subscribe")]
+    assert subscribed, "it subscribed to nothing"
+    assert published, "it published nothing"
+    assert max(subscribed) < min(
+        published
+    ), f"a row went out before the subscriptions were live: {bus.calls}"

tests/scheduler/test_scheduler.py

@@ -8,7 +8,7 @@
 from tc49.lib.layout import Layout
 from tc49.lib.scenario import RequestSpec, TrainSpec
 from tc49.scheduler import Scheduler
-from tests.harness import load, retaining
+from tests.harness import Recording, load, retaining, subscribed_before_publishing
 
 
 def yard() -> Layout:
@@ -910,22 +910,6 @@ def test_an_answer_that_cannot_be_read_leaves_the_request_in_flight() -> None:
     assert facing(seen)["express_2"] == "up_e.A-to-B"
 
 
-class Recording(InProcessBus):
-    """An in-process bus that remembers the order it was called in."""
-
-    def __init__(self) -> None:
-        super().__init__(Clock())
-        self.calls: list[str] = []
-
-    def subscribe(self, topic_filter: str, handler: object) -> None:  # type: ignore[override]
-        self.calls.append(f"subscribe {topic_filter}")
-        super().subscribe(topic_filter, cast(object, handler))  # type: ignore[arg-type]
-
-    def publish(self, topic: str, payload: Payload) -> None:
-        self.calls.append(f"publish {topic}")
-        super().publish(topic, payload)
-
-
 def test_it_subscribes_before_it_publishes_anything() -> None:
     """The opening rows go out with the subscriptions already live.
 
@@ -939,11 +923,4 @@ def test_it_subscribes_before_it_publishes_anything() -> None:
     """
     bus = Recording()
     Scheduler(bus, yard(), seeded(), TIMETABLE)
-
-    published = [i for i, call in enumerate(bus.calls) if call.startswith("publish")]
-    subscribed = [i for i, call in enumerate(bus.calls) if call.startswith("subscribe")]
-    assert subscribed, "the scheduler subscribed to nothing"
-    assert published, "the scheduler published nothing"
-    assert max(subscribed) < min(
-        published
-    ), f"a row went out before the subscriptions were live: {bus.calls}"
+    subscribed_before_publishing(bus)