From 524e3eaab452d92dd6efe8d36b77bba5b288e134 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Thu, 13 Aug 2026 00:29:36 -0700 Subject: [PATCH 1/2] fix: serialize a property's publish under a per-property lock The publish-on-change memo added in 0.20.0 could record an older payload than the one that actually reached the wire last, after which the gate suppressed the publish that would have corrected the broker. Two threads reach the sequence: the application thread via set_value(), and the MQTT loop thread via on_connect -> refresh_tree(force=True) -> Node.publish() -> publish_value(force=True). Interleaved, the loop thread could send the old payload, the application thread could then send and memoize the new one, and the loop thread could finally overwrite the memo with the older payload it had sent first. The property then believed the broker held a value it did not, and a later set_value() of that value was skipped, so the wrong retained value persisted until the next genuine change or reconnect. compute-payload / publish / memoize is now one atomic unit per property. The lock is REENTRANT because set_value() takes it and calls publish_value(), which on the retraction path calls clear_value(); a plain Lock self-deadlocks on the commonest call in the SDK. It is per-property, so it never serializes a tree walk, and no path holds two, so there is no ordering hazard. It is deliberately held across the transport's publish(): releasing earlier reopens the window it exists to close. That is safe for the paho transport the SDK ships, and the code records why, because it is not obvious: paho invokes on_connect holding only _in_callback_mutex and takes _out_message_mutex only after the callback returns (sequential, not nested), and the one place the publish path touches _in_callback_mutex uses a non-blocking acquire(False) that threaded mode skips. A bring-your-own transport could still build that cycle by holding its own lock across the on-connect handler it wires to refresh_tree() while also requiring it in publish(); it must not. get_last_published_value() now reads the memo tuple once into a local rather than testing the attribute and subscripting it. The GIL makes that window practically unreachable, but a free-threaded build removes the accident, and CI covers 3.13. CI jobs that run code gain timeout-minutes: 5. A lock regression deadlocks whichever thread reaches it, and unbounded, GitHub would let that run to its six-hour default instead of reporting a failure. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/lint.yml | 1 + .github/workflows/publish.yml | 2 + .github/workflows/test.yml | 5 + CHANGELOG.md | 13 +- src/ebus_sdk/__init__.py | 2 +- src/ebus_sdk/homie.py | 275 ++++++++++++++++++++-------------- tests/test_homie_device.py | 78 ++++++++++ 7 files changed, 261 insertions(+), 115 deletions(-) diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 6981199..f5d889f 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -8,6 +8,7 @@ on: jobs: ruff: runs-on: ubuntu-latest + timeout-minutes: 5 steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 2c2664c..48a6d24 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -12,6 +12,8 @@ permissions: jobs: test: runs-on: ubuntu-latest + # The publish gate: a hang here must fail, not stall the release. + timeout-minutes: 5 steps: - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index efa382e..2df79bd 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -9,6 +9,11 @@ jobs: pytest: name: pytest (py${{ matrix.python-version }}) runs-on: ubuntu-latest + # The suite runs in ~2s, so anything approaching this is a hang, not slowness. + # Since 0.20.1 homie.Property takes a lock, and a lock regression (notably making + # it non-reentrant) deadlocks the thread that hits it: without a bound, GitHub + # would let that run to the 6-hour default rather than reporting a failure. + timeout-minutes: 5 strategy: # Run every version even if one fails, so a single-version break is # visible as exactly that rather than masking the rest. diff --git a/CHANGELOG.md b/CHANGELOG.md index 2259a3a..5661788 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to `ebus-sdk` are recorded here. Format follows [Keep a Chan ## [Unreleased] +## [0.20.1] — 2026-08-13 + +### Changed + +- CI: the `pytest` and `ruff` jobs, and the publish workflow's test gate, now carry `timeout-minutes: 5`. The suite runs in about two seconds, so anything near that bound is a hang rather than slowness. This matters from this release on: `homie.Property` now takes a lock, and a lock regression deadlocks whichever thread reaches it (making it non-reentrant deadlocks the main thread at the first `set_value`, which is most of the suite). Unbounded, GitHub would let that run to its six-hour default instead of reporting a failure, and a hung job reports nothing useful. The publish and release jobs are deliberately left unbounded, since a slow PyPI upload is not the same kind of event. + +### Fixed + +- `Property` now serializes "compute the payload, publish it, record what was published" under a per-property reentrant lock, so the publish-on-change memo can never disagree with the last write that actually reached the wire. Two threads reach that sequence: the application thread via `set_value()`, and the MQTT loop thread via `on_connect` → `refresh_tree(force=True)`. Interleaved, the loop thread could publish the old payload, the application thread could then publish and memoize the new one, and the loop thread could finally overwrite the memo with the older payload it had sent first. The property then believed the broker held a value it did not, and the 0.20.0 gate suppressed the very publish that would have corrected it, so the wrong retained value persisted until the next genuine change or reconnect. The window is narrow (it needs a reconnect refresh concurrent with a value update) and the class has never had a lock, so `_value` and `_ever_published` were already exposed to it in kind; 0.20.0 made the consequence durable rather than transient, which is what moves this from a latent wart to a fix. The lock is reentrant because `set_value()` calls `publish_value()` calls `clear_value()`, each taking it; a plain `Lock` self-deadlocks on the commonest call in the SDK. It is per-property, so it never serializes a tree walk, and no path holds two, so there is no ordering hazard. It is deliberately held across the transport's `publish()`: releasing earlier reopens the window it exists to close. No API change. ([#50](https://github.com/electrification-bus/python-sdk/issues/50)) + ## [0.20.0] — 2026-08-12 ### Added @@ -314,7 +324,8 @@ The 0.2.0 release introduces first-class parent/child device trees on both the d Initial public release on PyPI. It predates this repo's tagging convention (the earliest tag is `v0.1.4`), so there is no `v0.1.2` tag to read; the published artifact on PyPI is the record of the surface that shipped. -[Unreleased]: https://github.com/electrification-bus/python-sdk/compare/v0.20.0...HEAD +[Unreleased]: https://github.com/electrification-bus/python-sdk/compare/v0.20.1...HEAD +[0.20.1]: https://github.com/electrification-bus/python-sdk/releases/tag/v0.20.1 [0.20.0]: https://github.com/electrification-bus/python-sdk/releases/tag/v0.20.0 [0.19.0]: https://github.com/electrification-bus/python-sdk/releases/tag/v0.19.0 [0.18.1]: https://github.com/electrification-bus/python-sdk/releases/tag/v0.18.1 diff --git a/src/ebus_sdk/__init__.py b/src/ebus_sdk/__init__.py index 0eaccec..bd5af55 100644 --- a/src/ebus_sdk/__init__.py +++ b/src/ebus_sdk/__init__.py @@ -81,7 +81,7 @@ # Structural types for a caller-supplied MQTT client from ebus_sdk.transport import MqttControllerTransport, MqttDeviceTransport, MqttTransport -__version__ = "0.20.0" +__version__ = "0.20.1" __all__ = [ # Homie classes diff --git a/src/ebus_sdk/homie.py b/src/ebus_sdk/homie.py index 2a9766f..aceef82 100644 --- a/src/ebus_sdk/homie.py +++ b/src/ebus_sdk/homie.py @@ -57,6 +57,7 @@ class StrEnum(str, Enum): from dataclasses import dataclass from functools import partial +from threading import RLock # from deprecated import deprecated from typing import Any, Callable, List, Optional, Type, Union @@ -530,10 +531,41 @@ def __init__( # because it is derived at publish time from the node/device ids, and # set_node()/set_device() are public: a reparented property must not have # its first publish on the new topic suppressed by the old topic's memo. - # Held as ONE tuple rather than two fields so the pair cannot tear when the - # MQTT loop thread and an application thread publish concurrently (a single - # attribute assignment is atomic; two are not). + # Held as ONE tuple so the pair is written in a single atomic assignment. self._last_published: Optional[tuple] = None + # Serializes "compute the payload, publish it, record what was published" so + # that triple is atomic per property. Two threads reach it: the application + # thread via set_value(), and the MQTT loop thread via on_connect -> + # refresh_tree() -> Node.publish() -> publish_value(force=True). Without this + # they can interleave so the memo records an OLDER payload than the one that + # actually went out last, and the GH #50 gate then suppresses the publish that + # would correct the broker, stranding the wrong retained value. + # + # REENTRANT, and not optionally so: set_value() takes this lock and then calls + # publish_value(), which takes it again, so a plain Lock self-deadlocks on the + # single most common call in the SDK. publish_value() -> clear_value() (the + # retraction path) nests the same way. Same reason GroupedPropertyDict uses an + # RLock. Verified by mutation: swapping RLock for Lock hangs set_value(). + # + # Per-property, so it never serializes a tree walk, and no code path holds two + # property locks at once, so there is no lock-ordering hazard here. + # + # It IS held across the transport's publish(), which is deliberate: releasing + # it earlier reopens the very window this closes. That is safe for the paho + # transport the SDK ships, and the reason is worth recording because it is not + # obvious. The dangerous shape would be an A-B/B-A cycle in which the network + # thread holds a transport lock while invoking on_connect (-> refresh_tree -> + # publish_value, which wants this lock) that publish() also needs. In paho + # 2.x it does not arise: _handle_connack invokes on_connect holding only + # _in_callback_mutex and acquires _out_message_mutex only AFTER the callback + # returns (the two blocks are sequential, not nested), and the one place the + # publish path touches _in_callback_mutex (_packet_queue) uses a NON-blocking + # acquire(False) that threaded mode skips entirely. + # + # A bring-your-own transport could still construct that cycle by holding its + # own lock across the on-connect handler it wires to refresh_tree() while + # requiring the same lock in publish(). Such a transport must not do that. + self._publish_lock = RLock() self._initial_value_was_none = value is None # Check for skip_initial_publish flag from dict self._skip_initial_publish = from_dict.get("skip_initial_publish", False) if from_dict else False @@ -615,9 +647,14 @@ def set_value(self, value: Any) -> bool: nothing failed, and the broker holds the value. Callers cannot distinguish suppressed from published from the return; use ``get_last_published_value()`` if you need the memo itself. + + Thread-safe: the value write and its publish are one atomic unit per property, + so a concurrent forced republish (the MQTT loop thread's reconnect refresh) + cannot land between them. """ - self._value = value - return self.publish_value() + with self._publish_lock: + self._value = value + return self.publish_value() def round(self) -> Optional[int]: """ @@ -814,83 +851,86 @@ def publish_value(self, *, force: bool = False) -> bool: reconnect would find every payload equal to what it "last published", send nothing, and leave those topics empty until each value happened to change. """ - mqttc = self.get_mqtt_client() - # Gate on connectivity, not just the SDK-owned run flag. A bring-your-own- - # transport client (mqttc=) is driven on the caller's loop and never has - # is_running set by the SDK's start(), yet can still publish once connected. - # is_running covers the owned path (True after start()); for an owned client - # connected implies running, so this does not change owned behavior. - if not mqttc or not (mqttc.is_running or mqttc.is_connected()): - _log_missing_client( - f"reason=propertyPublishValueNoMqttClient,id={self._id}", by_design=self._transport_free() - ) - return False - node_id = self.get_node_id() - device_id = self.get_device_id() - if not (device_id and node_id): - logger.warning( - f"propertyPublishValueInsufficientIDs,deviceID={device_id},nodeID={node_id},propertyID={self._id}" - ) - return False - # FIX: Don't publish if value is None and we've never published before or skip flag is set - if self._value is None and (not self._ever_published or self._skip_initial_publish): - logger.debug(f"reason=propertySkipPublishNoneValue,propertyID={self._id}") - return True - topic = f"{EBUS_HOMIE_DOMAIN}/{EBUS_HOMIE_VERSION_MAJOR}/{device_id}/{node_id}/{self._id}" - if self._value is None: - # Value was cleared after having been published. Emit the empty - # retained message so the prior retained value is retracted from the - # broker rather than silently left behind (a reconnecting subscriber - # would otherwise read the stale value). Reaching here implies - # _ever_published is True and _skip_initial_publish is False — the - # earlier guard returns for the never-published / skip-initial case. - # - # NOTE: this clears the topic (empty MQTT payload). It does NOT - # represent an actual empty-string *value*, which the Homie 5 - # convention encodes as a 1-character 0x00 payload — see the module - # header "empty string values" note. That encoding IS implemented, - # below, via encode_empty_string(); the two payloads are distinct and - # only the zero-length one retracts a retained topic. - logger.debug( - f"reason=propertyPublishValueIsNoneClearing,deviceID={device_id},nodeID={node_id},propertyID={self._id}" - ) - return self.clear_value() - try: - value = self.coerced_value() - if value is None: + # Serialize compute-publish-memoize so the memo can never record an older + # payload than the one that actually went out last (see _publish_lock). + with self._publish_lock: + mqttc = self.get_mqtt_client() + # Gate on connectivity, not just the SDK-owned run flag. A bring-your-own- + # transport client (mqttc=) is driven on the caller's loop and never has + # is_running set by the SDK's start(), yet can still publish once connected. + # is_running covers the owned path (True after start()); for an owned client + # connected implies running, so this does not change owned behavior. + if not mqttc or not (mqttc.is_running or mqttc.is_connected()): + _log_missing_client( + f"reason=propertyPublishValueNoMqttClient,id={self._id}", by_design=self._transport_free() + ) + return False + node_id = self.get_node_id() + device_id = self.get_device_id() + if not (device_id and node_id): logger.warning( - f"reason=propertyPublishValueCoercionFailed,propertyID={self._id},rawValue={self._value}" + f"propertyPublishValueInsufficientIDs,deviceID={device_id},nodeID={node_id},propertyID={self._id}" ) return False - # Encode an empty-string value as a single 0x00 byte so the broker - # does not mistake it for a zero-length "clear retained" payload. - payload = encode_empty_string(value) - # GH #50: skip a republish whose final wire payload is byte-identical to - # the one already sitting on this topic. Compared AFTER coercion and - # empty-string encoding, so the rounding/enum/JSON collapse is inside the - # comparison and an empty-string value ("\x00") can never alias the - # zero-length "clear retained" payload. - # - # RETAINED only. The broker stores nothing for an event property, so an - # identical consecutive payload there is a second real event and dropping - # it would lose information rather than save a redundant write. - # Truthiness rather than `is True`: retained is Optional[bool] and may be - # None, which the publish call below already treats as non-retained. - if not force and self.retained() and self._ever_published and self._last_published == (topic, payload): - logger.debug(f"reason=propertyPublishValueUnchanged,propertyID={self._id},topic={topic}") + # FIX: Don't publish if value is None and we've never published before or skip flag is set + if self._value is None and (not self._ever_published or self._skip_initial_publish): + logger.debug(f"reason=propertySkipPublishNoneValue,propertyID={self._id}") return True - logger.debug(f"reason=propertyPublishValue,value={value},topic={topic},retained={self.retained()}") - mqttc.publish(topic, payload, retain=self.retained(), qos=self._qos) - self._ever_published = True # FIX: Mark as published - # Memoize only after publish() returns, inside the try: a transport that - # raises must not leave a memo claiming the broker holds a payload it - # never received, which would suppress that value forever. - self._last_published = (topic, payload) - self._skip_initial_publish = False # FIX: Clear skip flag after first publish - return True - except Exception as e: - logger.warning(f"reason=propertyPublishValuePublishException,e={e}") - return False + topic = f"{EBUS_HOMIE_DOMAIN}/{EBUS_HOMIE_VERSION_MAJOR}/{device_id}/{node_id}/{self._id}" + if self._value is None: + # Value was cleared after having been published. Emit the empty + # retained message so the prior retained value is retracted from the + # broker rather than silently left behind (a reconnecting subscriber + # would otherwise read the stale value). Reaching here implies + # _ever_published is True and _skip_initial_publish is False — the + # earlier guard returns for the never-published / skip-initial case. + # + # NOTE: this clears the topic (empty MQTT payload). It does NOT + # represent an actual empty-string *value*, which the Homie 5 + # convention encodes as a 1-character 0x00 payload — see the module + # header "empty string values" note. That encoding IS implemented, + # below, via encode_empty_string(); the two payloads are distinct and + # only the zero-length one retracts a retained topic. + logger.debug( + f"reason=propertyPublishValueIsNoneClearing,deviceID={device_id},nodeID={node_id},propertyID={self._id}" + ) + return self.clear_value() + try: + value = self.coerced_value() + if value is None: + logger.warning( + f"reason=propertyPublishValueCoercionFailed,propertyID={self._id},rawValue={self._value}" + ) + return False + # Encode an empty-string value as a single 0x00 byte so the broker + # does not mistake it for a zero-length "clear retained" payload. + payload = encode_empty_string(value) + # GH #50: skip a republish whose final wire payload is byte-identical to + # the one already sitting on this topic. Compared AFTER coercion and + # empty-string encoding, so the rounding/enum/JSON collapse is inside the + # comparison and an empty-string value ("\x00") can never alias the + # zero-length "clear retained" payload. + # + # RETAINED only. The broker stores nothing for an event property, so an + # identical consecutive payload there is a second real event and dropping + # it would lose information rather than save a redundant write. + # Truthiness rather than `is True`: retained is Optional[bool] and may be + # None, which the publish call below already treats as non-retained. + if not force and self.retained() and self._ever_published and self._last_published == (topic, payload): + logger.debug(f"reason=propertyPublishValueUnchanged,propertyID={self._id},topic={topic}") + return True + logger.debug(f"reason=propertyPublishValue,value={value},topic={topic},retained={self.retained()}") + mqttc.publish(topic, payload, retain=self.retained(), qos=self._qos) + self._ever_published = True # FIX: Mark as published + # Memoize only after publish() returns, inside the try: a transport that + # raises must not leave a memo claiming the broker holds a payload it + # never received, which would suppress that value forever. + self._last_published = (topic, payload) + self._skip_initial_publish = False # FIX: Clear skip flag after first publish + return True + except Exception as e: + logger.warning(f"reason=propertyPublishValuePublishException,e={e}") + return False def clear_value(self) -> bool: """ @@ -915,40 +955,43 @@ def clear_value(self) -> bool: holds what that memo claims: re-setting the pre-retraction value afterwards republishes rather than being skipped. """ - # FIX: Don't clear if we never published a value - # This prevents creating phantom topics during cleanup - if not self._ever_published: - logger.info(f"reason=propertySkipClearNeverPublished,propertyID={self._id}") - return True + # Same lock as publish_value (reentrant: publish_value delegates here on + # the retraction path), so the retract and the memo reset are atomic. + with self._publish_lock: + # FIX: Don't clear if we never published a value + # This prevents creating phantom topics during cleanup + if not self._ever_published: + logger.info(f"reason=propertySkipClearNeverPublished,propertyID={self._id}") + return True - mqttc = self.get_mqtt_client() - # See publish_value: gate on connectivity so an injected (caller-driven) - # client can retract a retained value even without the SDK's is_running. - if not mqttc or not (mqttc.is_running or mqttc.is_connected()): - _log_missing_client( - f"reason=propertyClearValueNoMqttClient,propertyID={self._id}", by_design=self._transport_free() - ) - return False - node_id = self.get_node_id() - device_id = self.get_device_id() - if not (device_id and node_id): - logger.warning( - f"reason=propertyClearValueInsufficientIDs,deviceID={device_id},nodeID={node_id},propertyID={self._id}" - ) - return False - topic = f"{EBUS_HOMIE_DOMAIN}/{EBUS_HOMIE_VERSION_MAJOR}/{device_id}/{node_id}/{self._id}" - try: - # Publishing empty string clears retained message - mqttc.publish(topic, "", retain=True, qos=self._qos) - logger.info(f"reason=propertyClearedValue,propertyID={self._id},topic={topic}") - self._ever_published = False # FIX: Reset the flag - # The memo means "the broker holds this payload on this topic"; the - # retraction above has just made that false (GH #50). - self._last_published = None - return True - except Exception as e: - logger.warning(f"reason=propertyClearValueException,propertyID={self._id},topic={topic},exception={e}") - return False + mqttc = self.get_mqtt_client() + # See publish_value: gate on connectivity so an injected (caller-driven) + # client can retract a retained value even without the SDK's is_running. + if not mqttc or not (mqttc.is_running or mqttc.is_connected()): + _log_missing_client( + f"reason=propertyClearValueNoMqttClient,propertyID={self._id}", by_design=self._transport_free() + ) + return False + node_id = self.get_node_id() + device_id = self.get_device_id() + if not (device_id and node_id): + logger.warning( + f"reason=propertyClearValueInsufficientIDs,deviceID={device_id},nodeID={node_id},propertyID={self._id}" + ) + return False + topic = f"{EBUS_HOMIE_DOMAIN}/{EBUS_HOMIE_VERSION_MAJOR}/{device_id}/{node_id}/{self._id}" + try: + # Publishing empty string clears retained message + mqttc.publish(topic, "", retain=True, qos=self._qos) + logger.info(f"reason=propertyClearedValue,propertyID={self._id},topic={topic}") + self._ever_published = False # FIX: Reset the flag + # The memo means "the broker holds this payload on this topic"; the + # retraction above has just made that false (GH #50). + self._last_published = None + return True + except Exception as e: + logger.warning(f"reason=propertyClearValueException,propertyID={self._id},topic={topic},exception={e}") + return False def was_ever_published(self) -> bool: """Return whether this property has ever been published to MQTT (FIX for MQTT topic persistence)""" @@ -967,7 +1010,8 @@ def invalidate_publish_cache(self) -> None: Does NOT touch ``_ever_published``: this says "I no longer know what the broker holds", not "I have never published". """ - self._last_published = None + with self._publish_lock: + self._last_published = None def get_last_published_value(self) -> Optional[str]: """Return the wire PAYLOAD this property last published, or None if it has @@ -978,7 +1022,12 @@ def get_last_published_value(self) -> Optional[str]: ``value()`` for that. Before 0.20.0 this returned the current value, which was a documented placeholder rather than a real record (GH #50). """ - return self._last_published[1] if self._last_published else None + # Read the tuple ONCE into a local. Testing the attribute and then subscripting + # it would load it twice, and a concurrent clear_value() / invalidate_publish_ + # cache() nulling it in between would raise TypeError. The GIL makes that window + # practically unreachable today; a free-threaded build removes that accident. + memo = self._last_published + return memo[1] if memo else None def description(self) -> dict: """ diff --git a/tests/test_homie_device.py b/tests/test_homie_device.py index 21d77f8..70757c9 100644 --- a/tests/test_homie_device.py +++ b/tests/test_homie_device.py @@ -2,6 +2,7 @@ import json import logging +import threading from enum import Enum from unittest.mock import MagicMock, patch @@ -682,6 +683,83 @@ def test_invalidate_publish_cache_forces_the_next_publish(self): mock_client.publish.assert_called_once() +class TestHomiePropertyPublishThreadSafety: + """The publish-on-change memo must never disagree with the last wire write. + + Two threads reach ``publish_value``: the application thread via ``set_value``, + and the MQTT loop thread via ``on_connect`` -> ``refresh_tree(force=True)``. + Compute-publish-memoize has to be atomic per property, or the memo can record + an OLDER payload than the one that actually went out last, after which the + GH #50 gate suppresses the very publish that would correct the broker. + """ + + def test_a_concurrent_forced_republish_cannot_leave_a_stale_memo(self): + mock_client = _mock_mqtt_client() + prop = _make_wired_property(mock_client, value=1.0) + + in_publish = threading.Event() + release = threading.Event() + first = {"seen": False} + + def blocking_publish(*args, **kwargs): + # Block only the FIRST publish, so the loop thread is parked inside the + # transport with its payload already computed. + if not first["seen"]: + first["seen"] = True + in_publish.set() + release.wait(timeout=5) + return MagicMock(rc=0) + + mock_client.publish.side_effect = blocking_publish + + # daemon=True on both: if a future regression genuinely deadlocks a property, + # these threads never finish. The is_alive() assertion below reports that, but + # non-daemon threads would then block interpreter exit and CI would hang after + # printing the failure (neither the workflows nor pytest set a timeout). + loop_thread = threading.Thread(target=lambda: prop.publish_value(force=True), daemon=True) + loop_thread.start() + assert in_publish.wait(timeout=5), "forced republish never reached the transport" + + # Application thread: a new value arrives mid-flight. + app_thread = threading.Thread(target=lambda: prop.set_value(2.0), daemon=True) + app_thread.start() + app_thread.join(timeout=0.2) # must NOT complete: it is waiting on the lock + + release.set() + loop_thread.join(timeout=5) + app_thread.join(timeout=5) + assert not loop_thread.is_alive() and not app_thread.is_alive() + + payloads = [c[0][1] for c in mock_client.publish.call_args_list] + # The invariant the lock buys: the memo is what the broker last received. + # Unlocked, the app thread's "2.0" lands first and the loop thread then + # overwrites the memo with the older "1.0", so a later set_value(1.0) is + # suppressed while the broker still holds 2.0, which is permanently wrong. + assert prop.get_last_published_value() == payloads[-1], ( + f"memo {prop.get_last_published_value()!r} disagrees with the wire {payloads!r}" + ) + assert payloads == ["1.0", "2.0"] + + def test_retraction_through_the_lock_does_not_deadlock(self): + # The lock must be REENTRANT. set_value() takes it and calls publish_value(), + # which takes it again; on the None path publish_value() then delegates to + # clear_value(), which takes it a third time. A plain Lock self-deadlocks. + mock_client = _mock_mqtt_client() + prop = _make_wired_property(mock_client) + prop.set_value(99.0) + + done = threading.Event() + + def retract(): + prop.set_value(None) + done.set() + + t = threading.Thread(target=retract, daemon=True) + t.start() + assert done.wait(timeout=5), "set_value(None) deadlocked on a non-reentrant lock" + assert mock_client.publish.call_args[0][1] == "" # the retraction reached the wire + + class TestHomiePropertyClearValue: def test_clear_value_never_published_skips(self): mock_client = _mock_mqtt_client() From 584d3c1c4e77534655d86f91972394233b3c19a1 Mon Sep 17 00:00:00 2001 From: Donald Clark Jackson Date: Thu, 13 Aug 2026 00:34:43 -0700 Subject: [PATCH 2/2] fix: make the bulk-update context per-thread BulkUpdateContext.__enter__ and __exit__ mutated a shared _bulk_mode / _bulk_context pair on GroupedPropertyDict with no lock held, while every other accessor on the class (including the observer dispatch and the group scans) takes its RLock. Two threads entering bulk contexts on the same dict corrupted each other: entering displaced the other's context, and whichever exited first cleared bulk mode for both. No events were lost, because __exit__ fires the context object's own list, but they were misattributed and fragmented. Some of one thread's changes landed in the other's batch, and everything after the early exit fired individually instead of batching. That fragmentation is the real cost for a Homie publisher: each structural event escaping the batch triggers its own $description republish and a $state transition, so one logical change produces extra republishes and visible state flapping. The active context is now thread-local rather than serialized on the existing lock, because a bulk context can be held across I/O and one thread should not block for the duration of another's batch. Two threads batching independently was always the reasonable reading of this API; now it is the behavior. A nested bulk_update() on the same thread now restores the enclosing context on exit instead of clearing it, so the outer batch resumes rather than leaking its remainder as individual events. Closes #55. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 2 + src/ebus_sdk/property.py | 35 ++++++++++++----- tests/test_property.py | 83 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5661788..e91d060 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ All notable changes to `ebus-sdk` are recorded here. Format follows [Keep a Chan ### Fixed +- `GroupedPropertyDict`'s active bulk-update context is now per-thread. `BulkUpdateContext.__enter__` and `__exit__` mutated a shared `_bulk_mode`/`_bulk_context` pair with no lock held, while every other accessor on the class (including the observer dispatch) took its `RLock`, which made the omission look accidental rather than deliberate. Two threads entering bulk contexts on the same dict therefore corrupted each other: entering displaced the other's context, and whichever exited first cleared bulk mode for both. No events were lost, since `__exit__` fires the context object's own list, but they were misattributed and fragmented: some of one thread's changes landed in the other's batch, and everything after the early exit fired individually instead of batching. For a Homie publisher that fragmentation is the real cost, because each structural event that escapes the batch triggers its own `$description` republish and a `$state` transition, so one logical change produces extra republishes and visible state flapping. Thread-local rather than serializing on the existing lock, since a bulk context can be held across I/O and one thread should not block for the duration of another's batch. A nested `bulk_update()` on the same thread now restores the enclosing context on exit instead of ending it, so the outer batch resumes rather than leaking its remainder as individual events. Two threads batching independently was always the reasonable reading; now it is the actual behavior. Reported with a precise account of which consequences do and do not follow. ([#55](https://github.com/electrification-bus/python-sdk/issues/55)) + - `Property` now serializes "compute the payload, publish it, record what was published" under a per-property reentrant lock, so the publish-on-change memo can never disagree with the last write that actually reached the wire. Two threads reach that sequence: the application thread via `set_value()`, and the MQTT loop thread via `on_connect` → `refresh_tree(force=True)`. Interleaved, the loop thread could publish the old payload, the application thread could then publish and memoize the new one, and the loop thread could finally overwrite the memo with the older payload it had sent first. The property then believed the broker held a value it did not, and the 0.20.0 gate suppressed the very publish that would have corrected it, so the wrong retained value persisted until the next genuine change or reconnect. The window is narrow (it needs a reconnect refresh concurrent with a value update) and the class has never had a lock, so `_value` and `_ever_published` were already exposed to it in kind; 0.20.0 made the consequence durable rather than transient, which is what moves this from a latent wart to a fix. The lock is reentrant because `set_value()` calls `publish_value()` calls `clear_value()`, each taking it; a plain `Lock` self-deadlocks on the commonest call in the SDK. It is per-property, so it never serializes a tree walk, and no path holds two, so there is no ordering hazard. It is deliberately held across the transport's `publish()`: releasing earlier reopens the window it exists to close. No API change. ([#50](https://github.com/electrification-bus/python-sdk/issues/50)) ## [0.20.0] — 2026-08-12 diff --git a/src/ebus_sdk/property.py b/src/ebus_sdk/property.py index 942790a..ab14055 100644 --- a/src/ebus_sdk/property.py +++ b/src/ebus_sdk/property.py @@ -16,7 +16,7 @@ import uuid import logging from enum import Enum -from threading import Lock, RLock +from threading import Lock, RLock, local from typing import List, Callable, Union, Optional, Any, Type @@ -210,16 +210,28 @@ class BulkUpdateContext: def __init__(self, grouped_dict: "GroupedPropertyDict") -> None: self.grouped_dict = grouped_dict + # Only ever appended to by the thread that entered this context, because the + # active context is thread-local, so this list needs no lock of its own. self.pending_events = [] + self._previous = None def __enter__(self): - self.grouped_dict._bulk_mode = True - self.grouped_dict._bulk_context = self + # The active context is per-thread (GH #55). Two threads batching updates to + # one GroupedPropertyDict are independent: previously both wrote the same pair + # of shared flags, so entering one context displaced the other and whichever + # exited first ended bulk mode for both, fragmenting the survivor's batch into + # individual events. Thread-local rather than serializing on the existing lock, + # because a bulk context can be held across I/O and one thread should not block + # for the duration of another's batch. + state = self.grouped_dict._bulk_state + # Restore rather than clear on exit, so a nested bulk_update() on this thread + # resumes the enclosing batch instead of ending it. + self._previous = getattr(state, "context", None) + state.context = self return self def __exit__(self, exc_type, exc_val, exc_tb): - self.grouped_dict._bulk_mode = False - self.grouped_dict._bulk_context = None + self.grouped_dict._bulk_state.context = self._previous if not exc_type and self.pending_events: # Fire single BULK_UPDATE event with all changes for observer_id, callback in self.grouped_dict._observers.items(): @@ -401,8 +413,9 @@ def __init__(self) -> None: self._lock = RLock() # RLock needed because _fire_event is called while holding lock self._groups = {} self._observers = {} - self._bulk_mode = False - self._bulk_context = None + # The bulk-update context in effect FOR THE CALLING THREAD, if any (GH #55). + # Replaces a shared _bulk_mode/_bulk_context pair that two threads corrupted. + self._bulk_state = local() def _get_group(self, group: str) -> Optional[PropertyDict]: """Get the PropertyDict for the given group name, or None if not found""" @@ -667,9 +680,13 @@ def bulk_update(self) -> BulkUpdateContext: def _fire_event(self, event_type: ChangeEvent, **kwargs): """Fire an event to all observers""" - if self._bulk_mode and self._bulk_context: + # Per-thread: a bulk context opened on another thread must not swallow this + # thread's events, and must not stop swallowing its own when that other thread + # exits (GH #55). + bulk_context = getattr(self._bulk_state, "context", None) + if bulk_context is not None: # In bulk mode, accumulate events - self._bulk_context.add_event(event_type, **kwargs) + bulk_context.add_event(event_type, **kwargs) else: # Fire immediately with self._lock: diff --git a/tests/test_property.py b/tests/test_property.py index ffca14b..7512a3c 100644 --- a/tests/test_property.py +++ b/tests/test_property.py @@ -1,5 +1,6 @@ """Tests for ebus_sdk.property (ObservableProperty, PropertyDict, GroupedPropertyDict).""" +import threading from unittest.mock import MagicMock @@ -245,6 +246,88 @@ def test_bulk_update(self): # Should get a single BULK_UPDATE, not individual PROPERTY_CHANGED assert ChangeEvent.BULK_UPDATE in events + def test_concurrent_bulk_updates_do_not_fragment_each_others_batches(self): + """GH #55: the active bulk context is per-thread. + + Sharing one _bulk_mode/_bulk_context pair meant a second thread entering a + context displaced the first, and whichever exited first ended bulk mode for + both. The survivor's remaining changes then escaped as individual events. For + a Homie publisher that is extra $description republishes and $state flapping. + """ + gpd = GroupedPropertyDict() + gpd.add_property("a", Property(id="x", value=0)) + gpd.add_property("b", Property(id="x", value=0)) + + events = [] + events_lock = threading.Lock() + + def observe(event_type, **kw): + with events_lock: + events.append((event_type, kw.get("changes"))) + + gpd.add_observer(observe) + + b_entered = threading.Event() + b_exited = threading.Event() + + def thread_a(): + with gpd.bulk_update(): + gpd.set_value("a", "x", 1) + b_entered.wait(timeout=5) # let B open its context inside ours... + b_exited.wait(timeout=5) # ...and close it, before we make our 2nd change + gpd.set_value("a", "x", 2) + + def thread_b(): + with gpd.bulk_update(): + gpd.set_value("b", "x", 1) + b_entered.set() + b_exited.set() + + ta = threading.Thread(target=thread_a, daemon=True) + tb = threading.Thread(target=thread_b, daemon=True) + ta.start() + tb.start() + ta.join(timeout=10) + tb.join(timeout=10) + assert not ta.is_alive() and not tb.is_alive() + + bulk = [changes for et, changes in events if et is ChangeEvent.BULK_UPDATE] + stray = [et for et, _ in events if et is not ChangeEvent.BULK_UPDATE] + + # Exactly two batches, and nothing escaped as an individual event. Unlocked, + # thread A's second set_value fires on its own because B's exit cleared the + # shared flag. + assert stray == [], f"changes escaped their batch: {stray}" + assert len(bulk) == 2, f"expected one batch per thread, got {len(bulk)}" + + # Each batch holds only its own thread's changes. + groups_per_batch = [{c.get("group_name") for c in changes} for changes in bulk] + assert {"a"} in groups_per_batch, groups_per_batch + assert {"b"} in groups_per_batch, groups_per_batch + a_batch = next(c for c in bulk if {x.get("group_name") for x in c} == {"a"}) + assert len(a_batch) == 2, f"thread A's batch was fragmented: {a_batch}" + + def test_a_nested_bulk_update_resumes_the_enclosing_batch(self): + # The context restores the previous one on exit rather than clearing it, so an + # inner bulk_update() no longer ends the outer batch and leaks the rest of it + # as individual events. + gpd = GroupedPropertyDict() + gpd.add_property("g", Property(id="x", value=0)) + + events = [] + gpd.add_observer(lambda event_type, **kw: events.append((event_type, kw.get("changes")))) + + with gpd.bulk_update(): + gpd.set_value("g", "x", 1) + with gpd.bulk_update(): + gpd.set_value("g", "x", 2) + gpd.set_value("g", "x", 3) # still batched: the outer context resumed + + stray = [et for et, _ in events if et is not ChangeEvent.BULK_UPDATE] + assert stray == [], f"a change escaped the resumed outer batch: {stray}" + batches = [changes for et, changes in events if et is ChangeEvent.BULK_UPDATE] + assert [len(b) for b in batches] == [1, 2], batches # inner flushes first, then outer + def test_has_group(self): gpd = GroupedPropertyDict() assert gpd.has_group("missing") is False