fix(schema-1): per-message cost no longer grows with the device tree

packages/schema-1/CHANGELOG.md

@@ -35,6 +35,8 @@ Requires `span-panel-api` **3.8.0 or newer**, whose snapshot fields accept the u
 - **The grid-forming device name falls back to vendor and model instead of showing a device id.**
 - **A panel with no MID reports `dsm_state` `DSM_ON_GRID` and `current_run_config` `PANEL_ON_GRID` whenever it is publishing**, where both dropped to `UNKNOWN` each time `power-flows/grid` read within 1 W of zero
   ([#215](https://github.com/SpanPanel/span-panel-api/issues/215)).
+- **Each MQTT message costs the same few microseconds however many devices the panel declares**, where routing it tested every recorded subscription in turn and every readiness check walked the device tree. On a 30-device tree the work per message falls
+  from 90–250 µs to about 2 µs, measured on a desktop machine ([#223](https://github.com/SpanPanel/span-panel-api/issues/223)).
 
 ## [1.2.2]
 

packages/schema-1/src/span_panel_api_schema_1/adapter.py

@@ -118,7 +118,14 @@ def __init__(
         self._controller = Controller(root_device_id=serial_number, mqttc=self._routes)
         self._property_callbacks: list[Callable[[str, str, str, str | None], None]] = []
         self._awaiting: tuple[str, ...] = ()
+        # Readiness as last computed, or None once the tree has changed since.
+        self._ready: bool | None = None
         self._controller.set_on_property_changed_callback(self._on_property_changed)
+        # Every event that can change what `is_ready()` answers; see there.
+        self._controller.set_on_device_discovered_callback(self._tree_changed)
+        self._controller.set_on_device_state_changed_callback(self._tree_changed)
+        self._controller.set_on_device_removed_callback(self._tree_changed)
+        self._controller.set_on_description_received_callback(self._tree_changed)
         # Records the subscriptions the tree walk needs; nothing reaches the
         # wire, because this object has no connection to reach it with.
         self._controller.start_discovery()
@@ -173,7 +180,22 @@ def is_ready(self) -> bool:
         panel's positions with none to lay out. Asking only for what the panel
         itself promised keeps a firmware that omits the property connectable —
         it falls back to the drift warning in `panel_size_from_model`.
+
+        **The answer is remembered until the tree changes.** The transport asks
+        before and after every message, and walking the tree each time cost
+        more than routing the message did (#223). Only the Controller's own
+        events can move it: a device's first `$state`, a change of state, a
+        removal, a `$description`, and the root's `info/model` value. Each
+        forgets the answer and the next call works it out again. The events
+        never work it out themselves, because a description is announced before
+        the SDK has reconciled the children it declares.
         """
+        if self._ready is None:
+            self._ready = self._tree_ready()
+        return self._ready
+
+    def _tree_ready(self) -> bool:
+        """What `is_ready()` answers, worked out from the tree as it stands."""
         root = self._controller.get_root(self._serial_number)
         if root is None or root.state != STATE_READY or not root.description:
             return False
@@ -626,5 +648,12 @@ def _on_property_changed(self, device_id: str, node_id: str, property_id: str, v
         protocol has no place for the previous value, and consumers that need
         one keep it themselves.
         """
+        # The one value readiness waits on.
+        if (device_id, node_id, property_id) == (self._serial_number, NODE_INFO, PROP_MODEL):
+            self._ready = None
         for callback in list(self._property_callbacks):
             callback(device_id, node_id, property_id, value)
+
+    def _tree_changed(self, *_event: object) -> None:
+        """Forget the remembered readiness. Accepts any of the Controller's tree events."""
+        self._ready = None

packages/schema-1/src/span_panel_api_schema_1/transport.py

@@ -38,10 +38,11 @@
 
 from __future__ import annotations
 
+import itertools
 import logging
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, NamedTuple
 
-from paho.mqtt.client import topic_matches_sub
+from paho.mqtt.matcher import MQTTMatcher
 
 from span_panel_api.models import ControlTarget
 from span_panel_api_schema_1.const import HOMIE_DOMAIN, HOMIE_VERSION
@@ -80,6 +81,13 @@ def control_target(device_id: str, node: str, prop: str) -> ControlTarget:
 MAX_HELD_MESSAGES = 4096
 
 
+class _Route(NamedTuple):
+    """A recorded callback, and when it was recorded, which decides an overlap."""
+
+    recorded: int
+    callback: Callable[[str, bytes], None]
+
+
 class ControllerRoutes:
     """Record `Controller`'s subscriptions and route messages back to them.
 
@@ -88,8 +96,11 @@ class ControllerRoutes:
     """
 
     def __init__(self) -> None:
-        # Insertion-ordered; dispatch walks it in reverse — see dispatch().
-        self._routes: dict[str, Callable[[str, bytes], None]] = {}
+        self._routes: dict[str, _Route] = {}
+        # The same patterns, indexed for matching: each maps to itself, and
+        # dispatch resolves the patterns that match through `_routes`.
+        self._matcher = MQTTMatcher()
+        self._recordings = itertools.count()
         # Last payload per topic that matched no route yet, keyed by topic so a
         # newer value supersedes an older one — the same last-value-wins rule
         # the broker applies to the retained message this stands in for.
@@ -132,27 +143,35 @@ def subscribe(self, sub: str, param: Any, qos: int = 1) -> None:  # pylint: disa
         the adapter's behalf, not of a route recorded in a dict.
         """
         # Pop before insert: assigning an existing key updates the value but
-        # keeps the key's original position, which would leave a re-registered
-        # pattern behind whatever was added after it in dispatch's match order.
+        # keeps the key's original position, and `routes` lists most recent last.
         self._routes.pop(sub, None)
-        self._routes[sub] = param
+        self._routes[sub] = _Route(next(self._recordings), param)
+        self._matcher[sub] = sub
         self._release(sub, param)
 
     def unsubscribe(self, sub: str) -> None:
         """Forget the callback for `sub`.
 
         The broad wire subscription stays. Messages for a device the SDK has
-        dropped simply stop matching a route, and dispatch discards them.
+        dropped simply stop matching a route, and dispatch holds them.
         """
-        self._routes.pop(sub, None)
+        # Guarded because the matcher raises for a pattern it never held, and
+        # forgetting one that was never recorded is harmless.
+        if self._routes.pop(sub, None) is not None:
+            del self._matcher[sub]
 
     # -- our side ----------------------------------------------------------
 
     def dispatch(self, topic: str, payload: str) -> None:
         """Deliver one message to the callback of the route that matches.
 
-        Walked most-recent-first, which is defensive rather than currently
-        required. Tree-rooted discovery subscribes four **device-scoped**
+        Matched through a prefix tree of the recorded patterns, so the cost
+        follows the topic's depth rather than the number of routes. Every
+        message takes this path, and a linear scan grew with each device the
+        panel declares, at four routes per device (#223).
+
+        The most recently recorded match wins, which is defensive rather than
+        currently required. Tree-rooted discovery subscribes four **device-scoped**
         patterns per device — `$state`, `$description`, `+/+`, `+/+/$target`
         (`Controller._subscribe_device_topics`) — which cannot overlap each
         other or another device's, so today exactly one route matches any topic.
@@ -172,11 +191,15 @@ def dispatch(self, topic: str, payload: str) -> None:
 
         The SDK hands callbacks `bytes`; the transport hands us `str`.
         """
-        for sub in reversed(self._routes):
-            if topic_matches_sub(sub, topic):
-                self._routes[sub](topic, payload.encode())
-                return
-        self._hold(topic, payload)
+        # The matcher is untyped and yields what it was given, the pattern
+        # string; `_routes` turns each back into a typed route. The list is
+        # complete before any callback runs, so a callback that subscribes
+        # changes a matcher nothing is still walking.
+        matched = [self._routes[sub] for sub in self._matcher.iter_match(topic)]
+        if not matched:
+            self._hold(topic, payload)
+            return
+        max(matched, key=lambda route: route.recorded).callback(topic, payload.encode())
 
     def _hold(self, topic: str, payload: str) -> None:
         """Keep a message for a route that has not been registered yet."""
@@ -201,8 +224,13 @@ def _release(self, sub: str, callback: Callable[[str, bytes], None]) -> None:
         in turn, so a whole tree unfolds from one root subscription. The
         candidate list is therefore taken up front and each entry re-checked,
         since a nested release may have claimed it already.
+
+        The pattern is indexed once and matched against every held topic,
+        rather than indexed afresh for each of them.
         """
-        for topic in [held for held in self._held if topic_matches_sub(sub, held)]:
+        pattern = MQTTMatcher()
+        pattern[sub] = sub
+        for topic in [held for held in self._held if next(pattern.iter_match(held), None) is not None]:
             payload = self._held.pop(topic, None)
             if payload is not None:
                 callback(topic, payload.encode())

pyproject.toml

@@ -255,6 +255,12 @@ explicit_package_bases = true
 # `tests/` on `sys.path`; the support packages the mypy hook checks resolve the same
 # way here.
 mypy_path = "src:tests"
+# paho ships `py.typed` but leaves its topic matcher unannotated, so every call
+# into it would be an untyped call. The schema-1 route table resolves what the
+# matcher yields through a typed dict at once, which keeps its `Any` from going
+# further; `warn_return_any` does not police that for you (a union or a
+# `yield from` passes it), so keep it to that shape.
+untyped_calls_exclude = ["paho.mqtt.matcher"]
 # Excluded by name for the reason `[tool.ruff]` gives above: the capture script is
 # checked like the library, and the hand-run tools written before that was expected
 # are not.

tests/test_schema_one_adapter.py

@@ -9,9 +9,11 @@
 
 import copy
 import json
+import random
 
 import pytest
 
+from ebus_sdk import Controller
 from reference_payloads.schema_one import RetainedTopicTree, parent_child_tree
 from span_panel_api.adapters import _derive_required_members
 from span_panel_api.models import V2HomieSchema
@@ -52,14 +54,28 @@ def _feed(
     appears; `test_children_before_the_panel` covers the unfriendly order,
     which a broker is equally entitled to replay.
     """
+    for topic, payload in _retained(device_ids, omit, tree):
+        adapter.handle_message(topic, payload)
+
+
+def _retained(
+    device_ids: list[str] | None = None,
+    omit: tuple[str, ...] = (),
+    tree: RetainedTopicTree = _TREE,
+) -> list[tuple[str, str]]:
+    """The messages `_feed` delivers, in the order it delivers them."""
+    messages: list[tuple[str, str]] = []
     for device_id in device_ids or [PANEL, *[d for d in tree if d != PANEL]]:
         topics = tree[device_id]
         prefix = f"ebus/5/{device_id}"
-        adapter.handle_message(f"{prefix}/$description", topics["$description"])
-        adapter.handle_message(f"{prefix}/$state", topics["$state"])
-        for topic, value in topics.items():
-            if not topic.startswith("$") and topic not in omit:
-                adapter.handle_message(f"{prefix}/{topic}", value)
+        messages.append((f"{prefix}/$description", topics["$description"]))
+        messages.append((f"{prefix}/$state", topics["$state"]))
+        messages.extend(
+            (f"{prefix}/{topic}", value)
+            for topic, value in topics.items()
+            if not topic.startswith("$") and topic not in omit
+        )
+    return messages
 
 
 @pytest.fixture(name="adapter")
@@ -212,6 +228,152 @@ def test_a_panel_that_declares_no_model_still_connects() -> None:
     assert adapter.build_snapshot().panel_size == 0
 
 
+# ---------------------------------------------------------------------------
+# Readiness is remembered between messages
+#
+# The transport asks `is_ready()` before and after every message, and walking
+# the tree each time cost more than routing the message did (#223). The answer
+# is kept until the tree changes. Each check below asks after every message,
+# as the transport does, and compares the answer with an adapter that has seen
+# the same messages and is asked for the first time.
+# ---------------------------------------------------------------------------
+
+
+def _readiness_after_each(messages: list[tuple[str, str]], *, from_message: int = 0) -> list[bool]:
+    """Readiness after each message past `from_message`, each checked against a fresh adapter."""
+    adapter = SchemaOneAdapter(PANEL, _schema())
+    answers: list[bool] = []
+    for count, (topic, payload) in enumerate(messages, start=1):
+        adapter.handle_message(topic, payload)
+        remembered = adapter.is_ready()
+        if count <= from_message:
+            continue
+        fresh = SchemaOneAdapter(PANEL, _schema())
+        for earlier in messages[:count]:
+            fresh.handle_message(*earlier)
+        assert remembered is fresh.is_ready(), f"readiness was stale after {topic} = {payload!r}"
+        answers.append(remembered)
+    return answers
+
+
+def _panel_description(**changes: object) -> str:
+    description = json.loads(_TREE[PANEL]["$description"])
+    description.update(changes)
+    return json.dumps(description)
+
+
+def _shuffled(seed: int) -> list[tuple[str, str]]:
+    messages = _retained()
+    random.Random(seed).shuffle(messages)
+    return messages
+
+
+@pytest.mark.parametrize(
+    "messages",
+    [
+        pytest.param(_retained(), id="panel-first"),
+        pytest.param(_retained([*[d for d in _TREE if d != PANEL], PANEL]), id="children-first"),
+        *(pytest.param(_shuffled(seed), id=f"shuffled-{seed}") for seed in (1, 2, 3)),
+    ],
+)
+def test_readiness_follows_every_message_of_a_replay(messages: list[tuple[str, str]]) -> None:
+    """A broker replays its retained store in any order, `$state` before
+    `$description` and children before parents included."""
+    answers = _readiness_after_each(messages)
+
+    assert answers[-1] is True
+
+
+def test_readiness_follows_the_panel_leaving_ready_and_returning() -> None:
+    base = _retained()
+    messages = [*base, (f"ebus/5/{PANEL}/$state", "init"), (f"ebus/5/{PANEL}/$state", "ready")]
+
+    assert _readiness_after_each(messages, from_message=len(base)) == [False, True]
+
+
+def test_readiness_follows_a_child_that_is_removed_and_returns() -> None:
+    """An empty retained `$state` removes a device; the panel still declares it."""
+    base = _retained()
+    prefix = f"ebus/5/{KITCHEN_CIRCUIT}"
+    messages = [
+        *base,
+        (f"{prefix}/$state", ""),
+        (f"{prefix}/$state", "ready"),
+        (f"{prefix}/$description", _TREE[KITCHEN_CIRCUIT]["$description"]),
+    ]
+
+    assert _readiness_after_each(messages, from_message=len(base)) == [False, False, True]
+
+
+def test_readiness_follows_the_panel_withdrawing_a_child_it_was_waiting_on() -> None:
+    """A device commissioned and decommissioned before it ever described itself."""
+    base = _retained()
+    children = json.loads(_TREE[PANEL]["$description"])["children"]
+    messages = [
+        *base,
+        (f"ebus/5/{PANEL}/$description", _panel_description(children=[*children, "circuit-38"])),
+        (f"ebus/5/{PANEL}/$description", _TREE[PANEL]["$description"]),
+    ]
+
+    assert _readiness_after_each(messages, from_message=len(base)) == [False, True]
+
+
+def test_readiness_follows_the_panel_redeclaring_its_model() -> None:
+    """Whether the model is waited for is itself read from the panel's description."""
+    base = _retained(omit=("info/model",))
+    description = json.loads(_TREE[PANEL]["$description"])
+    del description["nodes"]["info"]["properties"]["model"]
+    messages = [
+        *base,
+        (f"ebus/5/{PANEL}/$description", json.dumps(description)),
+        (f"ebus/5/{PANEL}/$description", _TREE[PANEL]["$description"]),
+        (f"ebus/5/{PANEL}/info/model", _TREE[PANEL]["info/model"]),
+    ]
+
+    assert _readiness_after_each(messages, from_message=len(base)) == [True, False, True]
+
+
+def test_readiness_follows_a_childless_panel_whose_state_arrives_last() -> None:
+    """The panel's first `$state` is the only change here, and nothing else
+    re-examines the tree after it."""
+    panel = {topic: value for topic, value in _TREE[PANEL].items() if not topic.startswith("$")}
+    prefix = f"ebus/5/{PANEL}"
+    messages = [
+        (f"{prefix}/$description", _panel_description(children=[])),
+        *((f"{prefix}/{topic}", value) for topic, value in panel.items()),
+        (f"{prefix}/$state", "ready"),
+    ]
+
+    answers = _readiness_after_each(messages)
+
+    assert answers[-1] is True
+    assert not any(answers[:-1])
+
+
+def test_a_reading_or_a_rename_does_not_walk_the_tree(adapter: SchemaOneAdapter, monkeypatch: pytest.MonkeyPatch) -> None:
+    """The per-message cost #223 measured: values change constantly and
+    readiness never depends on them."""
+    assert adapter.is_ready() is True
+    walks = 0
+    is_tree_complete = Controller.is_tree_complete
+
+    def counting(controller: Controller, root_id: str) -> bool:
+        nonlocal walks
+        walks += 1
+        return is_tree_complete(controller, root_id)
+
+    monkeypatch.setattr(Controller, "is_tree_complete", counting)
+    prefix = f"ebus/5/{KITCHEN_CIRCUIT}"
+
+    adapter.handle_message(f"{prefix}/meter/active-power", "-250.0")
+    assert adapter.is_ready() is True
+    adapter.handle_message(f"{prefix}/info/name", "Kitchen island")
+    assert adapter.is_ready() is True
+
+    assert walks == 0
+    assert adapter.build_snapshot().circuits[KITCHEN_CIRCUIT].name == "Kitchen island"
+
+
 def test_snapshot_is_built_from_the_discovered_tree(adapter: SchemaOneAdapter) -> None:
     snapshot = adapter.build_snapshot()