Repository navigation
Controller: pace per-device subscriptions and heal stuck devices (#97) - #98
Merged
Merged
Conversation
A tree-rooted Controller subscribed to every child of a tree at once. The broker queues every matching retained message for that one client and, past its per-client limit (mosquitto max_queued_messages 1000), drops the rest; retained messages are sent only at subscribe time, so the dropped children's $state and $description never arrived. Prevent: per-device subscriptions (tree-rooted descendants and wildcard discovery) go through a queue of at most subscription_batch_size (default 8) in-flight devices. The next pending device is subscribed when an in-flight one has both $state and $description, is removed, or is declared stuck. Descendants are subscribed in the order their parent's $description lists them. None or 0 restores the unpaced fan-out. Heal: check_stuck_children(timeout=None) unsubscribes and resubscribes the filters of any watched device still lacking $state or $description after the timeout (default stuck_device_timeout, 10 s), up to max_resubscribe_attempts (default 3), logging each attempt and the final give-up. Message handling runs it at most once a second. No threads or timers; bookkeeping is under a lock and transport calls are made outside it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
…#97) Starts mosquitto with default limits on a free 127.0.0.1 port, seeds a retained tree (root plus 30 children, each with 60 property values, $state and $description) with a raw paho publisher, and asserts a tree-rooted Controller describes every child. Against main's src 17 of 30 children are described; with paced subscriptions all 30 are, in under a second. Skipped when no mosquitto binary is found. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
…by values (#97) Review fixes for the paced-subscription work: - Subscribe each device's $state and $description at once and its property filters only after both arrive. A declared child that never publishes, or a leftover $state with no $description, now holds no batch slot, so discovery no longer stalls behind it on a quiet connection. - Hold a slot until the property burst drains: every declared retained value arrived, or the $description resubscribed behind the property filters arrived again. Bound a batch by subscription_batch_values (default 500) as well as by device count. - A stuck slot holder rejoins the end of the queue instead of being resubscribed at once, so healing stays within the batch. - Always unsubscribe a dropped descendant, unsubscribe a $state for an untracked device in tree-rooted mode, and undo a subscribe that lands after the device was dropped on another thread. The real-broker tests now assert every property value and cover 150 and 300 values per child, wildcard discovery, and silent or $state-only devices. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
#97) - A device that gave up in stage one and publishes later is queued for its property filters (_note_device_progress queues any described device whose nodes lack property filters). - An injected mqttc is unpaced unless subscription_batch_size is passed: the drain marker needs a resubscribe to replay retained messages. - The value budget counts a value and a possible $target per retained property, and a slot is held until the marker or every value and target. - A device over the budget on its own is subscribed node by node in chunks that fit; a slot holder whose retry made no progress gives up; an owned client runs the stuck check from a timer thread. - $description is subscribed with_retain when the transport supports it, so a live republish is not taken for the marker; a changed description updates an active slot's expected count. - CHANGELOG/README/doc: tree-ready precedes descendants' values; untracked $state in tree-rooted mode is ignored and unsubscribed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
…ces (#97) On disconnect, a client the Controller built unsubscribes every discovered device's own filters, so its reconnect recovery resubscribes only the wildcard discovery filter or the tree root's filters. Wildcard resync() then restarts stage one for each device left without filters, waiting for a fresh $description before pacing its property filters; tree-rooted resync() already re-walks. Values changed during an outage now all arrive, and a large tree no longer stalls for stuck_device_timeout after a reconnect. A tree device that clears its $state keeps its $state/$description filters and drops its property filters. When it returns while still declared, or its parent declares it again, it is tracked again. _unsubscribe_and_drop now prunes and unsubscribes a device already gone from the registry. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
None keeps the default (thread only for a client the SDK builds), False never starts it, True starts it for an injected mqttc. stop() now joins the thread. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
…#97) 6c2a12d unsubscribed a removed tree device's property filters, the root included. resync() records the root's property filters as subscribed and the returning root takes the already-known branch, so after a reconnect or resync() the root's values never came back. Only descendants lose their property filters now, and the root behaves as on main. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #97.
A
Controllerthat subscribed to every device at once overflowed the broker's per-client queue (mosquittomax_queued_messages, default 1000), and the broker dropped the excess retained$state,$descriptionand property values for good. On a 30-child tree with 60 retained values per child, 17 of 30 children were described.Prevent: paced, staged subscriptions
$stateand$descriptionare subscribed as soon as it is known, so a declared child that never publishes holds nothing up.subscription_batch_sizedevices (default 8) andsubscription_batch_valuesretained messages (default 500, counting a value and a possible$targetper property). A device larger than the budget is subscribed node by node.mqttcstays unpaced unless the caller passessubscription_batch_size, because pacing relies on a resubscribe replaying retained messages.Heal: stuck devices
check_stuck_children(timeout=None)resubscribes a device still missing data afterstuck_device_timeout(default 10 s), up tomax_resubscribe_attempts(default 3), logging each attempt.ebus-controller-stuck-checkruns it on a quiet connection;stuck_check_threadchooses:None(default) only for a client the Controller builds,Falsenever,Truealways.stuck_device_timeout=Noneturns healing off.Behavior changes
on_tree_ready/is_tree_complete()fire before descendants' retained property values, which follow throughon_property_changed.$statefrom a device the tree does not declare is ignored and its filters unsubscribed. A descendant that clears its$stateand returns is discovered again.resync()re-paces every device left without filters.All are in CHANGELOG
[Unreleased].Tests
tests/test_controller_real_broker.pyruns 11 cases against a real mosquitto with default limits: trees of 60, 150, 300 and 1500 values per child, wildcard discovery, retained$target, quiet buses with silent devices, and values changed during a broker outage in both modes. Run againstmain's source, the ten that use default options all fail; the eleventh passesstuck_device_timeout, whichmainlacks. CI now installs mosquitto; the file skips when no binary is found.Known, low severity
After a reconnect, a tree root that has cleared its
$stateand does not return is treated as stuck: each of its 3 retries replays its retained values (identical values, extra callbacks), then it is logged as given up.🤖 Generated with Claude Code
https://claude.ai/code/session_01T32pvgqCzp1qCz7K5gS9Wm