From a2fe441a9b1d9ff0f3d47174813489e7bf37f393 Mon Sep 17 00:00:00 2001 From: cayossarian <23534755+cayossarian@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:03:23 -0700 Subject: [PATCH 1/2] fix(transport): route each message through one topic matcher ControllerRoutes.dispatch tested every recorded subscription in turn with paho's topic_matches_sub, which builds a fresh matcher for each call. Tree-rooted discovery records four routes per device, so every message cost grew with the panel, and the panel's own topics, recorded first and matched last, cost the most. The recorded patterns now live in one paho MQTTMatcher, and dispatch resolves the patterns that match through the typed route table. The most recently recorded match still wins, decided by a recording number rather than by the matcher's order, and a pattern is removed from the matcher only when it was recorded. _release indexes the new pattern once instead of once per held topic. paho leaves its matcher unannotated inside a py.typed package, so mypy exempts calls into paho.mqtt.matcher. What the matcher yields is used only as a key into the typed table. The new overlap test is the only one a first-match matcher fails; the existing ordering tests cannot tell recency from specificity. Refs #223 --- .../src/span_panel_api_schema_1/transport.py | 62 +++++++++++----- pyproject.toml | 6 ++ tests/test_schema_one_transport.py | 72 +++++++++++++++++++ 3 files changed, 123 insertions(+), 17 deletions(-) diff --git a/packages/schema-1/src/span_panel_api_schema_1/transport.py b/packages/schema-1/src/span_panel_api_schema_1/transport.py index 839f6a2f..827442c1 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/transport.py +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index caab61ce..e1bbb3c0 100644 --- a/pyproject.toml +++ b/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. diff --git a/tests/test_schema_one_transport.py b/tests/test_schema_one_transport.py index 52cecebe..f5f1296d 100644 --- a/tests/test_schema_one_transport.py +++ b/tests/test_schema_one_transport.py @@ -147,6 +147,78 @@ def test_rerecording_a_pattern_replaces_it_and_moves_it_to_most_recent() -> None first.assert_not_called() +def test_recency_decides_an_overlap_rather_than_specificity() -> None: + """The tests above cannot tell the two rules apart. + + In each of them the more recent route is also the more specific one, so a + matcher that simply returns the most specific filter passes them. Recorded + the other way round, only recency gives the broad route the message. + """ + routes = ControllerRoutes() + narrow = MagicMock(name="device") + broad = MagicMock(name="wildcard") + routes.subscribe("ebus/5/child-a/$state", narrow) + routes.subscribe("ebus/5/+/$state", broad) + + routes.dispatch("ebus/5/child-a/$state", "ready") + + broad.assert_called_once() + narrow.assert_not_called() + + +def test_unsubscribing_a_route_leaves_its_sibling_routing() -> None: + """Routes that share a prefix share it in the matcher too, so removing + one must leave the other where it was.""" + routes = ControllerRoutes() + state = MagicMock(name="state") + description = MagicMock(name="description") + routes.subscribe("ebus/5/child-a/$state", state) + routes.subscribe("ebus/5/child-a/$description", description) + + routes.unsubscribe("ebus/5/child-a/$description") + routes.dispatch("ebus/5/child-a/$state", "ready") + routes.dispatch("ebus/5/child-a/$description", "{}") + + state.assert_called_once_with("ebus/5/child-a/$state", b"ready") + description.assert_not_called() + assert routes.held == 1 + + +def test_unsubscribing_a_route_leaves_a_longer_one_through_it_routing() -> None: + """The SDK's own shape: a device's `+/+/$target` extends its `+/+`.""" + routes = ControllerRoutes() + values = MagicMock(name="values") + targets = MagicMock(name="targets") + routes.subscribe("ebus/5/child-a/+/+", values) + routes.subscribe("ebus/5/child-a/+/+/$target", targets) + + routes.unsubscribe("ebus/5/child-a/+/+") + routes.dispatch("ebus/5/child-a/switch/relay/$target", "OPEN") + routes.dispatch("ebus/5/child-a/meter/active-power", "1.0") + + targets.assert_called_once_with("ebus/5/child-a/switch/relay/$target", b"OPEN") + values.assert_not_called() + assert routes.held == 1 + + +def test_a_route_recorded_again_after_it_was_dropped_routes_again() -> None: + """A device the SDK drops and later re-adopts: what arrived in between is + held, released to the returning route, and later messages reach it.""" + routes = ControllerRoutes() + callback = MagicMock() + routes.subscribe("ebus/5/child-a/+/+", callback) + routes.unsubscribe("ebus/5/child-a/+/+") + routes.dispatch("ebus/5/child-a/meter/active-power", "1.0") + + routes.subscribe("ebus/5/child-a/+/+", callback) + routes.dispatch("ebus/5/child-a/meter/active-power", "2.0") + + assert [call.args for call in callback.call_args_list] == [ + ("ebus/5/child-a/meter/active-power", b"1.0"), + ("ebus/5/child-a/meter/active-power", b"2.0"), + ] + + # --------------------------------------------------------------------------- # Arrival order # From cd494da26ff021c2f09a0810426d8f1b87a87b85 Mon Sep 17 00:00:00 2001 From: cayossarian <23534755+cayossarian@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:04:18 -0700 Subject: [PATCH 2/2] fix(adapter): remember readiness until the tree changes The transport asks is_ready() before and after every message, and each call walked the declared tree twice, through _awaiting_descriptions and Controller.is_tree_complete. Readiness cannot change on an ordinary value, so on a quiet tree that work bought nothing. The adapter now keeps its last answer and forgets it on the Controller's own events: a device's first $state, a change of state, a removal, a $description, and the panel's info/model value. The next call works it out again, never the event itself, because a description is announced before the SDK has reconciled the children it declares. The walk is unchanged, moved to _tree_ready(). Each readiness test asks after every message, as the transport does, and compares with a fresh adapter fed the same messages. The replays cover five arrival orders, and the scenarios cover the panel leaving ready, a child removed and returning, a child declared and withdrawn, the model's declaration changing, and a childless panel whose $state arrives last. Disabling any one of the five events fails at least one of them; the discovered event is caught only by the childless panel. Refs #223 --- packages/schema-1/CHANGELOG.md | 2 + .../src/span_panel_api_schema_1/adapter.py | 29 +++ tests/test_schema_one_adapter.py | 172 +++++++++++++++++- 3 files changed, 198 insertions(+), 5 deletions(-) diff --git a/packages/schema-1/CHANGELOG.md b/packages/schema-1/CHANGELOG.md index f13cb0df..681241dc 100644 --- a/packages/schema-1/CHANGELOG.md +++ b/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] diff --git a/packages/schema-1/src/span_panel_api_schema_1/adapter.py b/packages/schema-1/src/span_panel_api_schema_1/adapter.py index 0ec1c941..f483f069 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/adapter.py +++ b/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 diff --git a/tests/test_schema_one_adapter.py b/tests/test_schema_one_adapter.py index 9f4f178f..46bc7b33 100644 --- a/tests/test_schema_one_adapter.py +++ b/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()