Controller: pace per-device subscriptions and heal stuck devices (#97)

.github/workflows/test.yml

@@ -29,5 +29,8 @@ jobs:
       - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
         with:
           python-version: ${{ matrix.python-version }}
+      # mosquitto for tests/test_controller_real_broker.py (GH #97), which starts its
+      # own broker on a free port and is skipped when the binary is absent.
+      - run: sudo apt-get update && sudo apt-get install -y mosquitto
       - run: pip install -e '.[dev]'
       - run: python -m pytest -q

CHANGELOG.md

@@ -34,10 +34,14 @@ Homie 5 compliance, phase 1: producer wire correctness ([#95](https://github.com
 - Encoders for `datetime` (`datetime`/`date` as ISO 8601, :132), `duration` (`timedelta` as `PTxHxMxS`, :137-144) and `color` (a component tuple in the first `format` color type, :118-123). These fell through to `str()`, which produced `2026-10-08 12:00:00+00:00`, `0:05:00` and `(255, 0, 0)`.
 - `Device.publish_alert(alert_id, message)`, `Device.clear_alert(alert_id)` and `Device.alerts()`. Alerts publish retained to `$alert/[alert ID]` (:517), are republished on reconnect, and are cleared by `delete()` (:289).
 - `Device.invalidate_description_cache()`, the `$description` counterpart of `Property.invalidate_publish_cache()`. `clear_retained_topic()` aimed at the device's own `$description` and `delete_all_from_mqtt()` call it, so a later `publish_description()` is no longer suppressed after the topic was cleared.
+- `Controller(subscription_batch_size=8, subscription_batch_values=500, stuck_device_timeout=10.0, max_resubscribe_attempts=3)` and `Controller.check_stuck_children(timeout=None)`. A device still lacking `$state` or `$description` after `timeout` seconds has those two filters unsubscribed and resubscribed, so the broker resends them; a device still holding a batch slot after `timeout` gives it up and rejoins the end of the queue. Each device is retried up to `max_resubscribe_attempts` times, with a `reason=stuckDeviceResubscribe` or `reason=stuckDeviceRequeue` warning per attempt and a `reason=stuckDeviceGaveUp` error at the end; a slot holder whose retry received no more than the attempt before it gives up then, with `noProgress=true`. A device that gave up lacking `$state` or `$description` and publishes them later gets its property filters then. Message handling runs the check at most once a second, and so does a daemon timer thread when the Controller built the client; with an injected client on a quiet connection, call it from your own timer. `Controller(stuck_check_thread=None)` keeps that default; `False` never starts the thread (message handling still runs the check) and `True` starts it for an injected `mqttc` too; `stop()` ends it. `stuck_device_timeout=None` turns off the automatic check and the thread ([#97](https://github.com/electrification-bus/python-sdk/issues/97)).
 
 ### Fixed
 
 - `delete()` followed by `stop()` re-created the deleted device as a bare `$state=disconnected` with no `$description` (:272, :288). After `delete()` the root's `stop()` announces nothing, flushes `delete()`'s retained clears and disconnects cleanly in either mode, and a reconnect's `refresh_tree()` republishes nothing.
+- A `Controller` could miss retained `$state`, `$description` and property values of some devices for good. It subscribed to every device at once, the broker queued every matching retained message for the one client, and past mosquitto's default per-client limit (`max_queued_messages` 1000) it dropped the rest; retained messages are sent only at subscribe time. On a 30-child tree with 60 retained values per child, 17 of 30 children were described. The Controller now subscribes each tree descendant's `$state` and `$description` (in wildcard discovery, its `$description`) as soon as the device is known, and its property filters once both have arrived, at most `subscription_batch_size` devices and `subscription_batch_values` retained messages at a time, counting a value and a possible `$target` per retained property. A device larger than `subscription_batch_values` on its own is subscribed node by node (`{node}/+` and `{node}/+/$target`) in chunks that fit. The next device starts when an earlier one's `$description`, resubscribed behind its property filters, arrives again (a live republish of `$description`, told apart by the retain flag from ebus-mqtt-client 0.7.0, does not count), or when every value and `$target` it can owe has arrived. A declared child that never publishes, or a leftover `$state` with no `$description`, holds no slot. All 30 children are described with every value at 60, 150 and 300 values per child, with every `$target` at 150 values and targets per child, and three children of 1500 values each get every value. Pacing is the default only for a client the Controller builds; an injected `mqttc` is unpaced unless `subscription_batch_size` is passed, because pacing needs a resubscribe to replay retained messages, which a transport that only records subscriptions does not do. `subscription_batch_size=None` restores the unpaced fan-out. The tree root and the single device of single-device mode are subscribed unpaced. A reconnect is paced too: when a client the Controller built disconnects, it unsubscribes every discovered device's own filters, so the client's reconnect recovery resubscribes only the wildcard discovery filter or the tree root's filters, and the devices are subscribed again as on first discovery. In wildcard mode `resync()` now does that for every device left without filters; in tree-rooted mode its re-walk already did. Values changed during an outage on a 30-device bus with 150 values each now all arrive; before, 24 of 30 devices kept their pre-outage values, and a large tree stalled for `stuck_device_timeout` after each reconnect ([#97](https://github.com/electrification-bus/python-sdk/issues/97)).
+- With pacing, a device's property values are not tracked until its `$description` arrives, and `on_tree_ready` / `is_tree_complete()` (every declared device described) come before descendants' retained property values, which arrive afterwards through `on_property_changed`. Unpaced, the subscribe order happened to deliver most of them first ([#97](https://github.com/electrification-bus/python-sdk/issues/97)).
+- In tree-rooted mode a `$state` for a device the tree does not declare (a late retained message on the filters of a dropped descendant) is ignored and that device's filters are unsubscribed; `on_device_discovered` no longer fires for it. It used to be registered as a new device. A tree descendant that clears its `$state` keeps its `$state` and `$description` filters and loses its property filters (the root keeps all of its filters); when it publishes `$state` again while still declared by its parent, or its parent declares it again, it is tracked and discovered again ([#97](https://github.com/electrification-bus/python-sdk/issues/97)).
 
 ## [0.24.0] — 2026-10-03
 

README.md

@@ -301,6 +301,25 @@ see the root and none of its children. As the publisher mutates the tree
 (`Device(parent=...)` to add, `child.delete()` to remove), descendants are
 subscribed or dropped on the parent's next init→ready transition.
 
+#### Paced subscriptions and stuck devices
+
+A broker queues every retained message matching a new subscription for that one client and drops what exceeds its per-client limit (mosquitto: `max_queued_messages` 1000). Retained messages are sent only at subscribe time, so a device whose `$state`, `$description` or property values were dropped never receives them. The Controller therefore subscribes each device in two stages (tree-rooted descendants and wildcard-discovered devices). Its `$state` and `$description` are subscribed as soon as the device is known. Once both have arrived, its property filters are subscribed, at most `subscription_batch_size` devices (default 8) and `subscription_batch_values` retained messages (default 500, counting a value and a possible `$target` for each retained property its `$description` declares) at a time. A device larger than `subscription_batch_values` on its own is subscribed node by node, in chunks that fit. A device keeps its slot until its `$description`, resubscribed behind its property filters, arrives again, or until every value and `$target` it can owe has arrived. A device that never publishes holds no slot, so it delays no other device, and if it publishes later it gets its property filters then.
+
+Pacing is on by default for a client the Controller builds from `mqtt_cfg` and off for an injected `mqttc`. Pass `subscription_batch_size` to pace an injected transport only if subscribing a filter again makes it deliver the retained messages again, as a broker connection does.
+
+`on_tree_ready` and `is_tree_complete()` mean every declared device has published its `$description`. With pacing, descendants' retained property values arrive after that, through `on_property_changed`.
+
+A device still lacking `$state` or `$description` after `stuck_device_timeout` seconds (default 10) has those two filters unsubscribed and resubscribed, so the broker resends them. A device still holding a slot after that long gives it up and rejoins the end of the queue. Each is retried up to `max_resubscribe_attempts` times (default 3); a slot holder whose retry received no more than the attempt before it is not retried again. Message handling runs this check at most once a second, and so does a daemon thread named `ebus-controller-stuck-check` when the Controller built the client. `stuck_check_thread=True` starts that thread for an injected `mqttc` too, and `stuck_check_thread=False` never starts it; `stop()` ends it. With an injected client on a connection that goes quiet and no thread, call it yourself:
+
+```python
+controller = Controller(mqtt_cfg=cfg, root_device_id='panel-1',
+                        subscription_batch_size=8, stuck_device_timeout=10.0)
+...
+healed = controller.check_stuck_children()  # or check_stuck_children(timeout=30)
+```
+
+`subscription_batch_size=None` subscribes all of every device's filters at once; `subscription_batch_values=None` counts devices only; `stuck_device_timeout=None` turns off the automatic check.
+
 ## Module Structure
 
 ```

doc/consuming-a-homie-tree.md

@@ -111,6 +111,8 @@ controller.set_on_tree_ready_callback(lambda root: snapshot(root.device_id))
 
 `is_tree_complete(root_id)` walks the declared tree and returns whether every device named transitively under it has published its own `$description`. It is safe to call at any time, as often as you like, and it will **flip back to False** when a device declares a new child. That is not a defect; it is the open child set showing through.
 
+Described means `$description`, not values. With the `Controller`'s default subscription pacing, each descendant's retained property values arrive after its `$description`, so on the tree-ready edge most of them may not have arrived yet. Read values from `on_property_changed`, or wait for the properties you need.
+
 `set_on_tree_ready_callback()` fires on the incomplete-to-complete edge and **re-arms**: a root that grows a new child fires again once that child describes itself. A tree commissioned in stages produces one call per settled shape rather than one call ever.
 
 Read that re-arming as a warning. If you take the first call as a barrier and stop listening, you have rebuilt the one-shot barrier from failure mode 1 out of the very API meant to prevent it. Handle every call.