From 455ed5b27c261171233ee167a6f9d80aa18e92d2 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 00:41:43 +0000 Subject: [PATCH 1/7] fix(config_flow): normalize a pasted PEM before parsing it A PEM pasted from a text editor can carry bytes cryptography's parser refuses outright: a UTF-8 BOM some Windows editors silently prepend, CRLF line endings, and a stray blank line a paste can introduce between the header/body/footer. None of those are meaningful in PEM, but any of them surfaces as an opaque InvalidHeader with no hint of what's wrong -- which is why the same certificate pasted from Command Prompt's `type` (no BOM, no stray blank lines) loads fine while the same file opened in an editor and copied doesn't (issue #291). Normalize at the point the pasted blob is first captured, not just before minting the leaf cert: the same string is stored in the config entry and reused to re-mint the leaf on a future reconfigure, so a raw copy would keep failing every time it's read back, not just on the first attempt. --- custom_components/localthings/config_flow.py | 30 +++++++++++++++-- tests/localthings/test_config_flow.py | 35 ++++++++++++++++++++ 2 files changed, 63 insertions(+), 2 deletions(-) diff --git a/custom_components/localthings/config_flow.py b/custom_components/localthings/config_flow.py index 5f06bdb..f9936c3 100644 --- a/custom_components/localthings/config_flow.py +++ b/custom_components/localthings/config_flow.py @@ -163,6 +163,26 @@ def _fetch_samsung_uuid() -> str: raise RuntimeError(f"UUID not found in {_SAMSUNG_CLOUD_HOST} certificate subject") +def _normalize_pem(text: str) -> str: + """Clean up a pasted PEM blob before handing it to `cryptography`'s + parser (issue #291). + + A PEM copied out of a text editor can carry a few bytes `cryptography` + refuses outright: a UTF-8 BOM some Windows editors silently prepend, CR + line endings (or CRLF), and blank lines a paste can introduce between + the header/body/footer. Any of those surfaces as an opaque + `InvalidHeader` with no hint of what's actually wrong -- which is why + the same certificate pasted from `type` (no BOM, no stray blank lines) + loads fine while the editor's copy doesn't. None of the stripped bytes + are meaningful in PEM: the format is BOM-free, line-ending-agnostic, and + has no blank-line syntax of its own. + """ + text = text.lstrip("\ufeff") + text = text.replace("\r\n", "\n").replace("\r", "\n") + lines = [line for line in text.split("\n") if line.strip()] + return "\n".join(lines) + + def _mint_leaf_cert(ca_cert_pem: str, ca_key_pem: str, uuid: str) -> tuple[str, str]: """Mint a fresh RSA-2048 leaf cert signed by the CA. @@ -722,8 +742,14 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): if leaf_cert and leaf_key: existing_leaf = (leaf_cert, leaf_key) else: - self._ca_cert_pem = user_input[CONF_CA_CERT_PEM].strip() - self._ca_key_pem = user_input[CONF_CA_KEY_PEM].strip() + # Normalized before the probe even runs (issue #291), not + # just before minting: the same pasted blob is also what + # gets stored in the config entry and reused to re-mint the + # leaf on a future reconfigure, so a raw copy with a BOM or + # CRLF line endings would keep failing every time it's read + # back, not just this once. + self._ca_cert_pem = _normalize_pem(user_input[CONF_CA_CERT_PEM]) + self._ca_key_pem = _normalize_pem(user_input[CONF_CA_KEY_PEM]) try: info = await self.hass.async_add_executor_job( diff --git a/tests/localthings/test_config_flow.py b/tests/localthings/test_config_flow.py index 466575c..55ddad2 100644 --- a/tests/localthings/test_config_flow.py +++ b/tests/localthings/test_config_flow.py @@ -75,6 +75,41 @@ async def test_successful_setup(hass: HomeAssistant, mock_probe) -> None: assert result["data"][CONF_CA_CERT_PEM] == MOCK_CA_CERT_PEM +async def test_setup_normalizes_messy_pasted_pem(hass: HomeAssistant, mock_probe) -> None: + """A PEM with a leading UTF-8 BOM, CRLF line endings, and a stray blank + line -- the kind a Windows text editor's copy produces, as opposed to a + `type` dump (issue #291) -- must still be accepted and stored in its + normalized form, not rejected with an opaque InvalidHeader.""" + messy_cert = "\ufeff" + MOCK_CA_CERT_PEM.replace("\n", "\r\n") + "\r\n\r\n" + messy_key = "\ufeff" + MOCK_CA_KEY_PEM.replace("\n", "\r\n") + + result = await hass.config_entries.flow.async_init(DOMAIN, context={"source": "user"}) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: MOCK_HOST, + CONF_CA_CERT_PEM: messy_cert, + CONF_CA_KEY_PEM: messy_key, + }, + ) + assert result["type"] == FlowResultType.CREATE_ENTRY + assert result["data"][CONF_CA_CERT_PEM] == MOCK_CA_CERT_PEM + assert result["data"][CONF_CA_KEY_PEM] == MOCK_CA_KEY_PEM + + +def test_normalize_pem_strips_bom_crlf_and_blank_lines() -> None: + """Unit-level check of the helper itself, isolated from the flow.""" + from custom_components.localthings.config_flow import _normalize_pem + + messy = "\ufeff-----BEGIN CERTIFICATE-----\r\nTEST-CA\r\n\r\n-----END CERTIFICATE-----\r\n" + assert _normalize_pem(messy) == ( + "-----BEGIN CERTIFICATE-----\nTEST-CA\n-----END CERTIFICATE-----" + ) + # A clean PEM (the `type`-dump case) passes through unchanged. + clean = "-----BEGIN CERTIFICATE-----\nTEST-CA\n-----END CERTIFICATE-----" + assert _normalize_pem(clean) == clean + + def test_order_candidates_prefers_known_ports() -> None: """Live ports are ordered with the historically known DTLS ports first, then the rest ascending.""" From 252306838db9f19a625ed584fe44b18a47fd08c5 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 00:42:31 +0000 Subject: [PATCH 2/7] fix(coordinator): downgrade observe mode when a device stays unreachable When a poll fails and the immediate reconnect retry fails too, _async_update_data returned the last-known snapshot as a degraded success (issue #254) without ever touching observe mode. That's fine for the data itself, but the connection-mode sensor reads straight from self._observe.mode, and only the *successful* reconnect branch ever changed it -- so a device that drops off the network entirely (air-gapped, powered off, Wi-Fi down) left that sensor reporting "Push" forever, hours after the session was actually dead (issue #287). Downgrade to poll mode on the failure branch too, without attempting an immediate resubscribe: the reconnect that would normally justify one just proved there's no live session to subscribe on. Recovery still happens on its own once the device is reachable again, via the existing poll-mode retry timer (_maybe_retry_observe_mode). --- custom_components/localthings/coordinator.py | 14 ++++++ tests/localthings/test_coordinator.py | 51 ++++++++++++++++++++ 2 files changed, 65 insertions(+) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index 64445f5..692fa22 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -834,6 +834,20 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): resources = await self.hass.async_add_executor_job(self._poll_once) except Exception as e2: self._log.error("poll failed after reconnect: %s", e2) + # The reconnect itself proved the old session is gone, so + # any OBSERVE subscription on it is too -- this cycle has + # no live session to resubscribe on, unlike the success + # branch below, so downgrade without setting + # just_downgraded_from_observe (that would attempt a + # resubscribe this same cycle against a session we just + # confirmed is unreachable). Without this, a device that + # drops off the network entirely leaves the connection-mode + # sensor reporting "Push" forever, since only the success + # path below ever changed it (issue #287) -- the poll-mode + # retry timer (_maybe_retry_observe_mode) still re-attempts + # observe once the device is actually reachable again. + if self._observe.mode == MODE_OBSERVE: + self._observe.downgrade_to_poll() snapshot = self._cache.snapshot() # Same precondition as _defer_reconnect_for (issue #254): # degraded-but-successful data only makes sense once diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 1711150..3338079 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -478,6 +478,57 @@ async def test_reconnect_while_observe_mode_downgrades_to_poll( assert coordinator._observe.mode == MODE_POLL +async def test_total_poll_failure_downgrades_observe_mode_to_poll( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session, fridge_resources +) -> None: + """A device that drops off the network entirely -- both the poll and its + reconnect retry fail -- must not leave the connection-mode sensor + reporting 'Push' forever (issue #287). Only the *successful* reconnect + branch used to touch observe mode (see + test_reconnect_while_observe_mode_downgrades_to_poll); this covers the + branch where the device stays unreachable.""" + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs + + fake.notify_on_subscribe = {"notified": True} + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, + hrefs, + 0.02, + 0.8, + ) + assert entered is True + assert coordinator.observe_mode == MODE_OBSERVE + + last_notify_ts = coordinator._observe._last_notify_ts + assert last_notify_ts is not None + coordinator._observe._last_notify_ts = last_notify_ts - (PUSH_HEALTH_WINDOW_S + 1) + + with ( + patch( + "custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once", + side_effect=[RuntimeError("connection lost"), RuntimeError("still lost")], + ), + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new=AsyncMock(), + ), + ): + await coordinator.async_request_refresh() + await hass.async_block_till_done() + + # The update still "succeeds" with the last-known snapshot (issue #254's + # degraded-data path) -- but the connection mode must reflect reality + # now, not the stale OBSERVE state from before the outage. + assert coordinator.observe_mode == MODE_POLL + assert coordinator.last_update_success is True + + async def test_poll_timeout_skips_reconnect_when_push_is_healthy( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: From 77c2d7831e95de73612bd38a5d5cc63a3d5bd7a6 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 00:44:35 +0000 Subject: [PATCH 3/7] fix(coordinator): retry a command once after a dead-session reconnect async_send_command's _do_put caught any exception, logged it, and returned -- no reconnect, no retry, no error the user could see. A command landing on a session Samsung's firmware closed between polls (the same 'known device behavior' _async_update_data already reconnects around) was silently lost, with nothing to do about it but a manual reload of the device (issue #294). Mirror the poll path's own recovery: on failure, close the dead session, pause, and retry the PUT once against a freshly reconnected one. If that also fails, raise a HomeAssistantError instead of just logging, so the user gets a visible error rather than a command that quietly did nothing. The retry runs under the same session lock the poll path uses, so a write landing mid-reconnect can't race a concurrent poll cycle rebuilding the same session. --- custom_components/localthings/coordinator.py | 38 +++++++-- .../localthings/translations/cs.json | 3 + .../localthings/translations/en.json | 3 + .../localthings/translations/es.json | 3 + .../localthings/translations/it.json | 3 + .../localthings/translations/ko.json | 3 + .../localthings/translations/nl.json | 3 + tests/localthings/test_coordinator.py | 83 +++++++++++++++++++ 8 files changed, 132 insertions(+), 7 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index 692fa22..c0484f6 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -15,7 +15,7 @@ from typing import Any import cbor2 from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback -from homeassistant.exceptions import ServiceValidationError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers import issue_registry as ir from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed @@ -1026,18 +1026,42 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): ) def _do_put(): + if self._session is None: + self._connect_session() sess = self._session if sess is None: raise RuntimeError("no session") code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S) self._log.info("PUT %s → code %#04x", write_href, code) - try: - await self.hass.async_add_executor_job(_do_put) - except Exception as e: - self._log.error("command failed for %s: %s", write_href, e) - else: - await self.async_request_refresh() + # A session Samsung's firmware closed between polls (the same "known + # device behavior" _async_update_data reconnects around) previously + # sank the PUT here with nothing but a log line -- no retry, no + # user-facing error, so the command was silently lost until the next + # manual reload (issue #294). Mirror the poll path's own recovery: + # one reconnect, one retry, and this time raise on final failure so + # the user gets a red toast instead of nothing. + # + # Locked the same as the poll path so a write landing mid-reconnect + # doesn't tear down a session _async_update_data is simultaneously + # rebuilding (or vice versa). + async with self._session_lock: + try: + await self.hass.async_add_executor_job(_do_put) + except Exception as e: + self._log.warning("command failed for %s, reconnecting: %s", write_href, e) + await self.hass.async_add_executor_job(self._close_session) + await asyncio.sleep(self._RECONNECT_PAUSE_S) + try: + await self.hass.async_add_executor_job(_do_put) + except Exception as e2: + self._log.error("command failed for %s after reconnect: %s", write_href, e2) + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="command_failed", + translation_placeholders={"href": write_href, "error": str(e2)}, + ) from e2 + await self.async_request_refresh() # ------------------------------------------------------------------ # Debug raw write (issue #54): a power-user escape hatch for the diff --git a/custom_components/localthings/translations/cs.json b/custom_components/localthings/translations/cs.json index 83e5b72..82ccf6f 100644 --- a/custom_components/localthings/translations/cs.json +++ b/custom_components/localthings/translations/cs.json @@ -1356,6 +1356,9 @@ }, "intensive_unavailable_for_cycle": { "message": "Intenzivní není u vybraného cyklu k dispozici." + }, + "command_failed": { + "message": "Příkaz pro {href} selhal i po opětovném připojení: {error}" } } } diff --git a/custom_components/localthings/translations/en.json b/custom_components/localthings/translations/en.json index d403e24..dbf61ee 100644 --- a/custom_components/localthings/translations/en.json +++ b/custom_components/localthings/translations/en.json @@ -1356,6 +1356,9 @@ }, "intensive_unavailable_for_cycle": { "message": "Intensive isn't available on the selected cycle." + }, + "command_failed": { + "message": "The command to {href} failed even after reconnecting: {error}" } } } diff --git a/custom_components/localthings/translations/es.json b/custom_components/localthings/translations/es.json index ec8e6a0..d24b5a8 100644 --- a/custom_components/localthings/translations/es.json +++ b/custom_components/localthings/translations/es.json @@ -119,6 +119,9 @@ }, "intensive_unavailable_for_cycle": { "message": "El modo intensivo no está disponible en el ciclo seleccionado." + }, + "command_failed": { + "message": "El comando para {href} falló incluso después de reconectar: {error}" } }, "entity": { diff --git a/custom_components/localthings/translations/it.json b/custom_components/localthings/translations/it.json index be69d02..8e3e0e7 100644 --- a/custom_components/localthings/translations/it.json +++ b/custom_components/localthings/translations/it.json @@ -1356,6 +1356,9 @@ }, "intensive_unavailable_for_cycle": { "message": "Intensivo non è disponibile per il ciclo selezionato." + }, + "command_failed": { + "message": "Il comando per {href} è fallito anche dopo la riconnessione: {error}" } } } diff --git a/custom_components/localthings/translations/ko.json b/custom_components/localthings/translations/ko.json index 7a86ba4..736928f 100644 --- a/custom_components/localthings/translations/ko.json +++ b/custom_components/localthings/translations/ko.json @@ -1356,6 +1356,9 @@ }, "intensive_unavailable_for_cycle": { "message": "선택한 코스에서는 강력 세탁을 사용할 수 없습니다." + }, + "command_failed": { + "message": "재연결 후에도 {href} 명령이 실패했습니다: {error}" } } } diff --git a/custom_components/localthings/translations/nl.json b/custom_components/localthings/translations/nl.json index 09e15db..2623d3b 100644 --- a/custom_components/localthings/translations/nl.json +++ b/custom_components/localthings/translations/nl.json @@ -1356,6 +1356,9 @@ }, "intensive_unavailable_for_cycle": { "message": "Intensief is niet beschikbaar voor het geselecteerde programma." + }, + "command_failed": { + "message": "Het commando naar {href} is ook na opnieuw verbinden mislukt: {error}" } } } diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 3338079..278e832 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -1007,6 +1007,89 @@ async def test_send_command_survives_stale_confirm_poll( assert coordinator._cache.get("/test/vs/0") == {"value": 5} +async def test_send_command_reconnects_and_retries_after_socket_closed( + hass: HomeAssistant, + mock_entry, + mock_coordinator_observe_session, +) -> None: + """A command lost to a session Samsung's firmware closed between polls + must not just vanish (issue #294): `_do_put` failing once is now + followed by a reconnect and a single retry, mirroring the poll path's + own recovery in `_async_update_data`.""" + from custom_components.localthings.registry.discovery import BoundEntity + from custom_components.localthings.registry.entities import NumberDesc + + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + def _write_fn(payload, rep, href=None): + return (["test", "vs", "0"], {"value": payload}) + + desc = NumberDesc(key="test", field="value", write_fn=_write_fn) + bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc) + + calls = {"n": 0} + + def _post(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise ConnectionError("socket closed") + return (0x44, b"") + + with ( + patch.object(fake, "subscribe"), + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new=AsyncMock(), + ), + ): + fake.post = _post + await coordinator.async_send_command(bound, 5) + + assert calls["n"] == 2 + assert coordinator._cache.get("/test/vs/0") == {"value": 5} + + +async def test_send_command_raises_after_reconnect_retry_also_fails( + hass: HomeAssistant, + mock_entry, + mock_coordinator_observe_session, +) -> None: + """If the command still fails on the reconnected session, the user must + see it -- previously this was swallowed into a log line with no + feedback at all (issue #294).""" + from homeassistant.exceptions import HomeAssistantError + + from custom_components.localthings.registry.discovery import BoundEntity + from custom_components.localthings.registry.entities import NumberDesc + + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + def _write_fn(payload, rep, href=None): + return (["test", "vs", "0"], {"value": payload}) + + desc = NumberDesc(key="test", field="value", write_fn=_write_fn) + bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc) + + def _post(*args, **kwargs): + raise ConnectionError("socket closed") + + with ( + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new=AsyncMock(), + ), + pytest.raises(HomeAssistantError), + ): + fake.post = _post + await coordinator.async_send_command(bound, 5) + + async def test_second_write_to_same_href_lands_during_first_writes_settle_window( hass: HomeAssistant, mock_entry, From 50bb8934076946fd1c0d6001bc6dd579b49a038b Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 01:06:22 +0000 Subject: [PATCH 4/7] review: re-arm the settle window on retry, tighten comments, close a test gap An Opus review of the three prior commits on this branch (PR #306) turned up two real defects and a documentation/test gap, all fixed here: - async_send_command's retry (issue #294) armed the write-settle window before the retry existed, so the reconnect pause plus a second PUT could eat into the time meant for the confirming poll, reviving the revert-then-reapply symptom the window was sized to prevent (issue #9). Re-arm it after a successful retry lands. - test_send_command_reconnects_and_retries_after_socket_closed relied on the observe-session fixture's no-op _close_session, so self._session never actually went None and _do_put's reconnect guard was never exercised -- the test passed even with that guard deleted. Now overrides _close_session/_connect_session to actually drop and rebuild the session, and asserts the reconnect happened. - async_send_command's docstring still said "Fire-and-forget", which stopped being true the moment it started retrying and raising. Also trimmed the three comment blocks the review flagged as reproducing their commit messages verbatim, per CONTRIBUTING.md's comment-style rules. One review finding is not addressed here and needs a decision: the new _close_session() call in the command-retry path isn't synchronized against _attempt_observe_mode, which touches the session without _session_lock. A write's reconnect can race an in-flight observe-mode subscribe attempt and tear down the session it's using. Fixing it properly means broadening lock scope around observe-mode entry, which risks blocking a write behind an up to ~15s subscribe grace period -- a tradeoff not made unilaterally here. A second finding (dropping the old .strip()'s per-line whitespace handling in _normalize_pem) did not reproduce against a real certificate/key, only against the test suite's placeholder PEM body, so it's left as-is. --- custom_components/localthings/config_flow.py | 25 +++--------- custom_components/localthings/coordinator.py | 43 +++++++++----------- tests/localthings/test_coordinator.py | 22 +++++++++- 3 files changed, 46 insertions(+), 44 deletions(-) diff --git a/custom_components/localthings/config_flow.py b/custom_components/localthings/config_flow.py index f9936c3..183d6d7 100644 --- a/custom_components/localthings/config_flow.py +++ b/custom_components/localthings/config_flow.py @@ -164,19 +164,10 @@ def _fetch_samsung_uuid() -> str: def _normalize_pem(text: str) -> str: - """Clean up a pasted PEM blob before handing it to `cryptography`'s - parser (issue #291). - - A PEM copied out of a text editor can carry a few bytes `cryptography` - refuses outright: a UTF-8 BOM some Windows editors silently prepend, CR - line endings (or CRLF), and blank lines a paste can introduce between - the header/body/footer. Any of those surfaces as an opaque - `InvalidHeader` with no hint of what's actually wrong -- which is why - the same certificate pasted from `type` (no BOM, no stray blank lines) - loads fine while the editor's copy doesn't. None of the stripped bytes - are meaningful in PEM: the format is BOM-free, line-ending-agnostic, and - has no blank-line syntax of its own. - """ + """Strip a pasted PEM's BOM, CRLF endings, and blank lines before + `cryptography` sees it -- a text editor's copy carries all three and + fails with an opaque InvalidHeader, while the same file dumped via + `type` doesn't (issue #291).""" text = text.lstrip("\ufeff") text = text.replace("\r\n", "\n").replace("\r", "\n") lines = [line for line in text.split("\n") if line.strip()] @@ -742,12 +733,8 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): if leaf_cert and leaf_key: existing_leaf = (leaf_cert, leaf_key) else: - # Normalized before the probe even runs (issue #291), not - # just before minting: the same pasted blob is also what - # gets stored in the config entry and reused to re-mint the - # leaf on a future reconfigure, so a raw copy with a BOM or - # CRLF line endings would keep failing every time it's read - # back, not just this once. + # Normalized here, not just before minting: this is also + # what gets stored and reused to re-mint the leaf later. self._ca_cert_pem = _normalize_pem(user_input[CONF_CA_CERT_PEM]) self._ca_key_pem = _normalize_pem(user_input[CONF_CA_KEY_PEM]) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index c0484f6..d747406 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -834,18 +834,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): resources = await self.hass.async_add_executor_job(self._poll_once) except Exception as e2: self._log.error("poll failed after reconnect: %s", e2) - # The reconnect itself proved the old session is gone, so - # any OBSERVE subscription on it is too -- this cycle has - # no live session to resubscribe on, unlike the success - # branch below, so downgrade without setting - # just_downgraded_from_observe (that would attempt a - # resubscribe this same cycle against a session we just - # confirmed is unreachable). Without this, a device that - # drops off the network entirely leaves the connection-mode - # sensor reporting "Push" forever, since only the success - # path below ever changed it (issue #287) -- the poll-mode - # retry timer (_maybe_retry_observe_mode) still re-attempts - # observe once the device is actually reachable again. + # Without this, a fully unreachable device left the + # connection-mode sensor stuck on "Push" forever -- only + # the success branch below ever downgraded it (issue + # #287). No just_downgraded_from_observe here: there's no + # live session this cycle to resubscribe on. if self._observe.mode == MODE_OBSERVE: self._observe.downgrade_to_poll() snapshot = self._cache.snapshot() @@ -924,7 +917,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): # ------------------------------------------------------------------ async def async_send_command(self, bound_entity: BoundEntity, payload: Any) -> None: - """Write a value to the device. Fire-and-forget. + """Write a value to the device. Retries once on a dead session + (issue #294); raises HomeAssistantError if that retry fails too. A description-level validate_fn (SwitchDesc only, currently) rejects a write with a user-facing message ahead of write_fn's silent @@ -1034,17 +1028,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S) self._log.info("PUT %s → code %#04x", write_href, code) - # A session Samsung's firmware closed between polls (the same "known - # device behavior" _async_update_data reconnects around) previously - # sank the PUT here with nothing but a log line -- no retry, no - # user-facing error, so the command was silently lost until the next - # manual reload (issue #294). Mirror the poll path's own recovery: - # one reconnect, one retry, and this time raise on final failure so - # the user gets a red toast instead of nothing. - # - # Locked the same as the poll path so a write landing mid-reconnect - # doesn't tear down a session _async_update_data is simultaneously - # rebuilding (or vice versa). + # Mirrors the poll path's reconnect-and-retry (issue #294): a PUT + # landing on a session Samsung's firmware closed between polls used + # to be silently lost -- no retry, no user-facing error. async with self._session_lock: try: await self.hass.async_add_executor_job(_do_put) @@ -1061,6 +1047,15 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): translation_key="command_failed", translation_placeholders={"href": write_href, "error": str(e2)}, ) from e2 + # The retry's own reconnect pause + second PUT can eat well + # into the settle window armed above, leaving too little of + # it for the confirming poll below and reviving the + # revert-then-reapply symptom settle_s exists to prevent + # (issue #9). Re-arm it fresh now that the write actually + # landed. + self._observe.mark_write_pending( + write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S + ) await self.async_request_refresh() # ------------------------------------------------------------------ diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 278e832..42cc435 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -1015,7 +1015,14 @@ async def test_send_command_reconnects_and_retries_after_socket_closed( """A command lost to a session Samsung's firmware closed between polls must not just vanish (issue #294): `_do_put` failing once is now followed by a reconnect and a single retry, mirroring the poll path's - own recovery in `_async_update_data`.""" + own recovery in `_async_update_data`. + + `mock_coordinator_observe_session` patches `_close_session` to a no-op, + which would leave `self._session` never actually going `None` -- and + with it, `_do_put`'s own `if self._session is None: self._connect_session()` + guard never exercised, so a broken reconnect could still pass. Overridden + here to actually drop the session, so the retry only succeeds if that + guard really rebuilds it.""" from custom_components.localthings.registry.discovery import BoundEntity from custom_components.localthings.registry.entities import NumberDesc @@ -1038,8 +1045,19 @@ async def test_send_command_reconnects_and_retries_after_socket_closed( raise ConnectionError("socket closed") return (0x44, b"") + def _drop_session(): + coordinator._session = None + + reconnects = {"n": 0} + + def _reconnect(): + reconnects["n"] += 1 + coordinator._session = fake + with ( patch.object(fake, "subscribe"), + patch.object(coordinator, "_close_session", side_effect=_drop_session), + patch.object(coordinator, "_connect_session", side_effect=_reconnect), patch( "custom_components.localthings.coordinator.asyncio.sleep", new=AsyncMock(), @@ -1048,6 +1066,8 @@ async def test_send_command_reconnects_and_retries_after_socket_closed( fake.post = _post await coordinator.async_send_command(bound, 5) + assert reconnects["n"] == 1 + assert calls["n"] == 2 assert coordinator._cache.get("/test/vs/0") == {"value": 5} From 69f93be4dca396447a2abfcce12ff5d6b612e820 Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 01:39:03 +0000 Subject: [PATCH 5/7] fix(coordinator): close the observe-mode race an Opus design review found The command-retry fix (issue #294) added a self._close_session() call to async_send_command that isn't synchronized against _attempt_observe_mode, which reads self._session and subscribes to it without holding _session_lock. A write's retry racing an in-flight subscribe attempt could tear down the session mid-subscribe -- or worse, land the close *after* the attempt's grace wait already succeeded, letting it commit observe mode against a session that's already gone: mode claims "Push" forever, with nothing left to notice the underlying socket is dead. Split ObserveManager.try_enter_observe_mode into four pieces (subscribe_hrefs / await_observe_notifies / enter_observe_mode / abandon_observe_attempt), keeping try_enter_observe_mode as a thin wrapper so its direct callers in test_observe.py are unaffected. _attempt_observe_mode now holds _session_lock only for the subscribe burst (each send is fire-and-forget, not a network round trip) and re-checks self._session is sess under the lock right before committing -- sess keeps the old session object alive, so identity can't be recycled onto a new one, which is what makes the check sufficient without a separate generation counter. The wait itself stays lock-free, so a command write is never blocked behind it. Two more bugs the same investigation turned up, fixed in the same pass since they're direct consequences of the design above: - async_send_command's own successful reconnect didn't downgrade observe mode the way the poll path's reconnect already does, leaving the same stale-commit problem reachable with zero concurrency at all -- just a write's retry succeeding while mode was observe. Replaced the poll path's local just_downgraded_from_observe with an instance flag both reconnect sites set, so either one triggers an immediate resubscribe. - _maybe_retry_observe_mode's 600s throttle gated solely on last_mode_change_ts, which _set_mode only stamps on an actual transition -- a device that never succeeds at observe mode leaves that timestamp stuck at construction time, so the throttle opens once and never closes again, re-attempting on every single poll cycle instead of every 600s. Now gates on the more recent of that timestamp and a new _last_observe_attempt_ts, stamped on every attempt regardless of outcome. --- custom_components/localthings/coordinator.py | 85 +++++++--- custom_components/localthings/observe.py | 80 +++++++--- tests/localthings/test_coordinator.py | 156 ++++++++++++++++++- 3 files changed, 279 insertions(+), 42 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index d747406..c8d6d32 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -204,6 +204,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._consecutive_poll_timeouts = 0 self._unbound_hrefs: list[str] = [] self._reconnect_times: list[float] = [] + # See _maybe_retry_observe_mode: last_mode_change_ts alone doesn't + # move on a failed attempt, so this tracks attempts too. + self._last_observe_attempt_ts = 0.0 + # Set by both reconnect paths (poll and command) that hand back a + # session with zero OBSERVE registrations while mode was still + # observe; consumed once to trigger an immediate resubscribe + # instead of waiting out _RECOVERY_RETRY_S. + self._resubscribe_due = False # ------------------------------------------------------------------ # Session management (all blocking — must run in executor) @@ -731,29 +739,62 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): ir.async_delete_issue(self.hass, DOMAIN, issue_id) async def _attempt_observe_mode(self) -> None: - """Called once, right after first discovery. Blocking (sleeps for - the whole grace period) — must run in an executor.""" + """Called once, right after first discovery, after a reconnect + downgrades from observe, and periodically while polling. Two + phases: subscribing holds `_session_lock` (each send is fire-and- + forget, not a network round trip); the grace wait that follows + does not, so a concurrent command write isn't blocked for the + whole ~15s wait (issue #294) -- only the brief subscribe burst. + + The wait can outlast a reconnect elsewhere (the poll path's own + recovery, or a command retry), which would otherwise let a stale + success commit observe mode against a session that's already been + replaced -- claiming "Push" with nothing left to notice it's dead. + `self._session is sess` re-checked under the lock right before + committing closes that: `sess` keeps the old object alive, so + identity can't be recycled onto a new one. + """ hrefs = self._hot_hrefs + self._warm_hrefs if not hrefs: return - if self._session is None: - # _poll_once already connects on a real poll; only fires if the - # session was closed out from under us concurrently. - await self.hass.async_add_executor_job(self._connect_session) - sess = self._session - if sess is None: + self._last_observe_attempt_ts = time.monotonic() + async with self._session_lock: + if self._session is None: + # _poll_once already connects on a real poll; only fires if + # the session was closed out from under us concurrently. + await self.hass.async_add_executor_job(self._connect_session) + sess = self._session + if sess is None: + return + subscribed = await self.hass.async_add_executor_job( + self._observe.subscribe_hrefs, sess, hrefs + ) + if not subscribed: + self._observe.abandon_observe_attempt() return - await self.hass.async_add_executor_job( - self._observe.try_enter_observe_mode, - sess, - hrefs, - self._OBSERVE_GRACE_PERIOD_S, + reached = await self.hass.async_add_executor_job( + self._observe.await_observe_notifies, subscribed, self._OBSERVE_GRACE_PERIOD_S ) + async with self._session_lock: + if not reached or self._session is not sess: + self._observe.abandon_observe_attempt() + return + self._observe.enter_observe_mode(sess, subscribed) async def _maybe_retry_observe_mode(self) -> None: """While in poll-only mode, periodically re-attempt observe mode - so a device that gains internet access recovers push automatically.""" - if time.monotonic() - self._observe.last_mode_change_ts < _RECOVERY_RETRY_S: + so a device that gains internet access recovers push automatically. + + Gated on the more recent of the two timestamps, not just + `last_mode_change_ts`: `_set_mode` only stamps that on an actual + transition, so a device that never successfully enters observe + mode would otherwise leave this throttle open forever after the + first `_RECOVERY_RETRY_S` window -- re-attempting (and paying the + subscribe-burst lock) on every single poll cycle instead of every + `_RECOVERY_RETRY_S`. + """ + last_attempt = max(self._observe.last_mode_change_ts, self._last_observe_attempt_ts) + if time.monotonic() - last_attempt < _RECOVERY_RETRY_S: return await self._attempt_observe_mode() @@ -805,7 +846,6 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._subpoll_task.cancel() self._subpoll_task = None - just_downgraded_from_observe = False async with self._session_lock: try: resources = await self.hass.async_add_executor_job(self._poll_once) @@ -862,7 +902,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): "poll and resubscribing on the new session" ) self._observe.downgrade_to_poll() - just_downgraded_from_observe = True + self._resubscribe_due = True if not self._discovered: # One-time (issue #177): find sibling subdevices before the @@ -895,7 +935,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): for href, rep in resources.items(): self._observe.apply(href, rep, source=source) - if first_cycle or just_downgraded_from_observe: + if first_cycle or self._resubscribe_due: + self._resubscribe_due = False await self._attempt_observe_mode() elif self._observe.mode == MODE_POLL: await self._maybe_retry_observe_mode() @@ -1056,6 +1097,14 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._observe.mark_write_pending( write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S ) + # A reconnect hands back a session with zero OBSERVE + # registrations, same as the poll path's own reconnect -- + # keeping observe mode here would leave those subscriptions + # claimed on the closed session with no poll failure left + # to notice (issue #294). + if self._observe.mode == MODE_OBSERVE: + self._observe.downgrade_to_poll() + self._resubscribe_due = True await self.async_request_refresh() # ------------------------------------------------------------------ diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index 3d91d51..04226cf 100644 --- a/custom_components/localthings/observe.py +++ b/custom_components/localthings/observe.py @@ -171,17 +171,16 @@ class ObserveManager: self._last_notify_ts is not None and time.monotonic() - self._last_notify_ts < window_s ) - def try_enter_observe_mode( - self, - session, - hrefs: list[str], - grace_period_s: float = GRACE_PERIOD_S, - success_fraction: float = SUCCESS_FRACTION, - ) -> bool: - """Blocking — subscribes to every href then waits up to - `grace_period_s`, returning early once `success_fraction` of hrefs - have notified. Caller must run this in an executor, never on the - event loop.""" + def subscribe_hrefs(self, session, hrefs: list[str]) -> set[str]: + """Register OBSERVE on every href; returns the ones that took. + Blocking — run in an executor. + + Split from the grace wait below (issue #294) so the coordinator can + hold its session lock for just these sends -- each is a fire-and- + forget UDP datagram (DtlsCoapSession.subscribe doesn't wait for the + device's ack), unlike the wait, which can block for the whole grace + period and must not hold a lock a command write is also waiting on. + """ with self._notify_cond: self._notified.clear() subscribed: set[str] = set() @@ -192,30 +191,65 @@ class ObserveManager: subscribed.add(href) except Exception as e: self.log.warning("subscribe %s failed: %s", href, e) + return subscribed + + def await_observe_notifies( + self, + subscribed: set[str], + grace_period_s: float = GRACE_PERIOD_S, + success_fraction: float = SUCCESS_FRACTION, + ) -> bool: + """Blocking — waits up to `grace_period_s`, returning early once + `success_fraction` of `subscribed` have notified. Touches no + session; safe to run without holding a session lock.""" if not subscribed: - self._stop_refresh_task() - self._set_mode(MODE_POLL) - self.subscribed_hrefs = set() return False def _fraction_reached() -> bool: return len(set(self._notified) & subscribed) / len(subscribed) >= success_fraction with self._notify_cond: - reached = self._notify_cond.wait_for( - _fraction_reached, - timeout=grace_period_s, - ) + return self._notify_cond.wait_for(_fraction_reached, timeout=grace_period_s) - if reached: - self.subscribed_hrefs = subscribed - self._set_mode(MODE_OBSERVE) - self.start_refresh_task(session) - return True + def enter_observe_mode(self, session, subscribed: set[str]) -> None: + """Commit a successful attempt. Caller must have re-confirmed + `session` is still the live one under its session lock (issue + #294) -- committing against a session a reconnect already replaced + would claim observe mode with nothing left to notice it's dead.""" + self.subscribed_hrefs = set(subscribed) + self._set_mode(MODE_OBSERVE) + self.start_refresh_task(session) + def abandon_observe_attempt(self) -> None: + """Drop a failed or stale attempt: no subscriptions worth keeping.""" self._stop_refresh_task() self.subscribed_hrefs = set() self._set_mode(MODE_POLL) + + def try_enter_observe_mode( + self, + session, + hrefs: list[str], + grace_period_s: float = GRACE_PERIOD_S, + success_fraction: float = SUCCESS_FRACTION, + ) -> bool: + """Blocking — subscribes to every href then waits up to + `grace_period_s`, returning early once `success_fraction` of hrefs + have notified. Caller must run this in an executor, never on the + event loop. + + Single-threaded convenience wrapper around the phase split above + (subscribe_hrefs / await_observe_notifies / enter_observe_mode / + abandon_observe_attempt) for callers -- direct and most existing + tests -- that don't need the lock-scoping those phases exist for.""" + subscribed = self.subscribe_hrefs(session, hrefs) + if not subscribed: + self.abandon_observe_attempt() + return False + if self.await_observe_notifies(subscribed, grace_period_s, success_fraction): + self.enter_observe_mode(session, subscribed) + return True + self.abandon_observe_attempt() return False def _set_mode(self, mode: str) -> None: diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 42cc435..131d3fa 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time from datetime import timedelta from unittest.mock import AsyncMock, patch @@ -21,6 +22,7 @@ from custom_components.localthings.const import ( SUMMARY_INTERVAL_S, ) from custom_components.localthings.coordinator import ( + _RECOVERY_RETRY_S, LocalThingsCoordinator, _local_source_port, ) @@ -30,7 +32,7 @@ from custom_components.localthings.registry.capabilities.common import ( remote_control_required_for_write, ) -from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL +from .conftest import ENTRY_DATA, MOCK_MODEL, MOCK_SERIAL, FakeObserveSession from .conftest import _load_fridge_resources as _load_fridge @@ -742,6 +744,95 @@ async def test_reconnect_from_observe_mode_resubscribes_immediately( assert coordinator.observe_mode == MODE_OBSERVE +async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """A reconnect (the poll path's own, or a command's retry) can swap + self._session while this attempt's grace wait is in flight -- it runs + without holding _session_lock precisely so a write isn't blocked behind + it (issue #294). Committing observe mode against the now-stale local + `sess` reference would claim "Push" on a session that's already gone, + with nothing left to notice -- the identity re-check under the lock + right before committing must catch this and abandon instead. + + Simulates the swap from inside await_observe_notifies itself rather + than via real concurrency: subscribe_hrefs (and its lock) has already + returned by the time that call runs, so this lands exactly in the + window the identity check exists to cover, deterministically.""" + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + other = FakeObserveSession() + + def _swap_session_mid_wait(subscribed, grace_period_s, success_fraction=None): + coordinator._session = other + return True + + with patch.object( + coordinator._observe, "await_observe_notifies", side_effect=_swap_session_mid_wait + ): + await coordinator._attempt_observe_mode() + + assert coordinator.observe_mode == MODE_POLL + assert coordinator._observe.subscribed_hrefs == set() + assert coordinator._observe._refresh_thread is None + + +async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """_set_mode only stamps last_mode_change_ts on an actual transition, + so a device that never successfully enters observe mode leaves that + timestamp stuck at construction time forever -- a failed attempt keeps + calling _set_mode(MODE_POLL) while already in MODE_POLL, a no-op. + Gating solely on that timestamp would make the 600s throttle open once + and then never close again, re-attempting (and paying the subscribe + burst) on every single poll cycle instead of every _RECOVERY_RETRY_S.""" + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + assert coordinator.observe_mode == MODE_POLL # never notified during setup + + # Simulate exactly the scenario above: mode_change_ts is old (as it + # would be forever, for a device that never gets push), but an attempt + # really did just run. + coordinator._observe.last_mode_change_ts = time.monotonic() - _RECOVERY_RETRY_S - 1 + coordinator._last_observe_attempt_ts = time.monotonic() + + with patch.object(fake, "subscribe") as mock_subscribe: + await coordinator._maybe_retry_observe_mode() + + mock_subscribe.assert_not_called() + + +async def test_attempt_observe_mode_releases_lock_before_the_grace_wait( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """The subscribe burst holds _session_lock (it touches the session); + the grace wait after it must not, or a command write could stall + behind up to _OBSERVE_GRACE_PERIOD_S of an unrelated observe-mode-entry + attempt (issue #294). By construction, subscribe_hrefs's own + `async with self._session_lock:` has already exited by the time + await_observe_notifies is even called -- checked here rather than + inferred from timing.""" + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + locked_during_wait = {"value": None} + + def _check_lock(subscribed, grace_period_s, success_fraction=None): + locked_during_wait["value"] = coordinator._session_lock.locked() + return True + + with patch.object(coordinator._observe, "await_observe_notifies", side_effect=_check_lock): + await coordinator._attempt_observe_mode() + + assert locked_during_wait["value"] is False + + async def test_sweep_mismatch_never_downgrades_a_live_observe_session( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -1110,6 +1201,69 @@ async def test_send_command_raises_after_reconnect_retry_also_fails( await coordinator.async_send_command(bound, 5) +async def test_send_command_reconnect_downgrades_observe_mode( + hass: HomeAssistant, + mock_entry, + mock_coordinator_observe_session, +) -> None: + """A command's own successful reconnect hands back a session with zero + OBSERVE registrations too, same as the poll path's reconnect -- must + downgrade the same way and flag a resubscribe, or observe mode stays + claimed against a session the write just replaced underneath it + (issue #294).""" + from custom_components.localthings.registry.discovery import BoundEntity + from custom_components.localthings.registry.entities import NumberDesc + + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs + + fake.notify_on_subscribe = {"notified": True} + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, + hrefs, + 0.02, + 0.8, + ) + assert entered is True + assert coordinator.observe_mode == MODE_OBSERVE + + def _write_fn(payload, rep, href=None): + return (["test", "vs", "0"], {"value": payload}) + + desc = NumberDesc(key="test", field="value", write_fn=_write_fn) + bound = BoundEntity(href="/test/vs/0", capability=coordinator.bound[0].capability, desc=desc) + + calls = {"n": 0} + + def _post(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise ConnectionError("socket closed") + return (0x44, b"") + + with ( + patch.object(fake, "subscribe") as mock_subscribe, + patch( + "custom_components.localthings.coordinator.asyncio.sleep", + new=AsyncMock(), + ), + ): + fake.post = _post + await coordinator.async_send_command(bound, 5) + + # _resubscribe_due is consumed by this same call's own trailing + # refresh (async_request_refresh is awaited, not fire-and-forget), + # so the visible effect is a resubscribe attempt, not a lingering + # flag value to assert on afterward. + assert mock_subscribe.called + + assert coordinator.observe_mode == MODE_POLL + + async def test_second_write_to_same_href_lands_during_first_writes_settle_window( hass: HomeAssistant, mock_entry, From a3cc918343cc4892bbe807501e79a2baf4a24a7a Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 02:01:30 +0000 Subject: [PATCH 6/7] fix(coordinator): two gaps a follow-up Opus review found in the split A second review of the observe-mode phase split (previous commit) found two real regressions it introduced, both in the same failure family it was built to close: - async_send_command's failed-retry branch closed the session, then raised without downgrading observe mode -- the downgrade only ran on the retry's success path. A retry that also fails still leaves the session dead, so mode was left claiming "Push" on a session that no longer exists, same as the bug this whole fix targets. Moved the downgrade to run right after the close, unconditionally on how the retry goes. - _attempt_observe_mode's stale-session abandon (the identity-check branch added in the previous commit) didn't flag a resubscribe. A session swap discovered there means a fresh, never-tried session now exists, but _last_observe_attempt_ts was already stamped for the now-abandoned attempt -- so that new session sat unsubscribed for up to _RECOVERY_RETRY_S (600s) instead of being retried on the next cycle. Now sets _resubscribe_due, same as the two reconnect paths do. Also closes two test-coverage gaps the same review surfaced by mutation testing: no test asserted the lock actually holds during the subscribe burst (only that it's released for the wait), and no test distinguished the max() in _maybe_retry_observe_mode's throttle from using _last_observe_attempt_ts alone -- both mutations left the full suite green. Added one test for each, plus extended two existing tests for the bug fixes above; all four confirmed via mutation testing (revert the fix, watch the new/extended test fail; restore it, watch it pass). One finding from the same review is intentionally left open: async_close is the one self._session writer that doesn't take _session_lock, so a close racing _attempt_observe_mode isn't covered by today's identity check. This is pre-existing (the lock didn't cover any of _attempt_observe_mode before this branch's earlier commits either), not a regression from this branch's work, and is a shutdown/unload-path question rather than the write-vs-observe-mode race this branch set out to fix. --- custom_components/localthings/coordinator.py | 29 +++++--- tests/localthings/test_coordinator.py | 76 +++++++++++++++++++- 2 files changed, 95 insertions(+), 10 deletions(-) diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index c8d6d32..45832de 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -776,8 +776,18 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._observe.await_observe_notifies, subscribed, self._OBSERVE_GRACE_PERIOD_S ) async with self._session_lock: - if not reached or self._session is not sess: + stale_session = self._session is not sess + if not reached or stale_session: self._observe.abandon_observe_attempt() + if stale_session: + # A reconnect elsewhere replaced the session while this + # attempt waited -- that session has never been tried, + # so retry it next cycle instead of leaving it + # unsubscribed for up to _RECOVERY_RETRY_S, which + # _last_observe_attempt_ts (already stamped above, for + # the now-abandoned session) would otherwise throttle + # for (issue #294). + self._resubscribe_due = True return self._observe.enter_observe_mode(sess, subscribed) @@ -1078,6 +1088,15 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): except Exception as e: self._log.warning("command failed for %s, reconnecting: %s", write_href, e) await self.hass.async_add_executor_job(self._close_session) + # The session is dead the moment it's closed, so any OBSERVE + # subscriptions on it are too -- downgrade here, before the + # retry, so a retry that also fails doesn't leave mode + # claiming "Push" on a session that no longer exists + # (issue #294; the poll path handles the same fact for its + # own reconnect the same way, unconditionally on close). + if self._observe.mode == MODE_OBSERVE: + self._observe.downgrade_to_poll() + self._resubscribe_due = True await asyncio.sleep(self._RECONNECT_PAUSE_S) try: await self.hass.async_add_executor_job(_do_put) @@ -1097,14 +1116,6 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._observe.mark_write_pending( write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S ) - # A reconnect hands back a session with zero OBSERVE - # registrations, same as the poll path's own reconnect -- - # keeping observe mode here would leave those subscriptions - # claimed on the closed session with no poll failure left - # to notice (issue #294). - if self._observe.mode == MODE_OBSERVE: - self._observe.downgrade_to_poll() - self._resubscribe_due = True await self.async_request_refresh() # ------------------------------------------------------------------ diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 131d3fa..e77da3b 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -755,6 +755,11 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( with nothing left to notice -- the identity re-check under the lock right before committing must catch this and abandon instead. + The new session is never-tried, though, not just abandoned: it must + flag an immediate resubscribe rather than let _last_observe_attempt_ts + (stamped for the now-abandoned attempt) throttle it for up to + _RECOVERY_RETRY_S. + Simulates the swap from inside await_observe_notifies itself rather than via real concurrency: subscribe_hrefs (and its lock) has already returned by the time that call runs, so this lands exactly in the @@ -762,6 +767,7 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + assert coordinator._resubscribe_due is False other = FakeObserveSession() @@ -777,6 +783,7 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( assert coordinator.observe_mode == MODE_POLL assert coordinator._observe.subscribed_hrefs == set() assert coordinator._observe._refresh_thread is None + assert coordinator._resubscribe_due is True async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_change( @@ -807,6 +814,30 @@ async def test_maybe_retry_observe_mode_uses_most_recent_attempt_not_just_mode_c mock_subscribe.assert_not_called() +async def test_maybe_retry_observe_mode_also_respects_a_mode_change_outside_an_attempt( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """The mirror of the case above: last_mode_change_ts can be the more + recent of the two as well, e.g. right after the poll or command path's + own downgrade (neither goes through _attempt_observe_mode, so neither + stamps _last_observe_attempt_ts). Dropping last_mode_change_ts from the + max() would let a device that was *just* downgraded get re-attempted + immediately instead of respecting _RECOVERY_RETRY_S.""" + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + assert coordinator.observe_mode == MODE_POLL + + coordinator._last_observe_attempt_ts = time.monotonic() - _RECOVERY_RETRY_S - 1 + coordinator._observe.last_mode_change_ts = time.monotonic() + + with patch.object(fake, "subscribe") as mock_subscribe: + await coordinator._maybe_retry_observe_mode() + + mock_subscribe.assert_not_called() + + async def test_attempt_observe_mode_releases_lock_before_the_grace_wait( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -833,6 +864,29 @@ async def test_attempt_observe_mode_releases_lock_before_the_grace_wait( assert locked_during_wait["value"] is False +async def test_attempt_observe_mode_holds_lock_during_the_subscribe_burst( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """The other half of the split above: the subscribe burst does touch + the session, so it must hold _session_lock -- that's what actually + stops a concurrent close from landing mid-subscribe (issue #294).""" + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + locked_during_subscribe = {"value": None} + real_subscribe_hrefs = coordinator._observe.subscribe_hrefs + + def _check_lock(session, hrefs): + locked_during_subscribe["value"] = coordinator._session_lock.locked() + return real_subscribe_hrefs(session, hrefs) + + with patch.object(coordinator._observe, "subscribe_hrefs", side_effect=_check_lock): + await coordinator._attempt_observe_mode() + + assert locked_during_subscribe["value"] is True + + async def test_sweep_mismatch_never_downgrades_a_live_observe_session( hass: HomeAssistant, mock_entry, mock_coordinator_observe_session ) -> None: @@ -1170,7 +1224,13 @@ async def test_send_command_raises_after_reconnect_retry_also_fails( ) -> None: """If the command still fails on the reconnected session, the user must see it -- previously this was swallowed into a log line with no - feedback at all (issue #294).""" + feedback at all (issue #294). + + Also covers a sibling bug the fix for that same issue introduced: the + session is closed the moment the first attempt fails, so any OBSERVE + subscriptions on it are already dead regardless of whether the retry + that follows succeeds -- a failed retry must still downgrade mode, or + it's left claiming "Push" on a session that no longer exists.""" from homeassistant.exceptions import HomeAssistantError from custom_components.localthings.registry.discovery import BoundEntity @@ -1180,6 +1240,18 @@ async def test_send_command_raises_after_reconnect_retry_also_fails( await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs + + fake.notify_on_subscribe = {"notified": True} + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, + hrefs, + 0.02, + 0.8, + ) + assert entered is True + assert coordinator.observe_mode == MODE_OBSERVE def _write_fn(payload, rep, href=None): return (["test", "vs", "0"], {"value": payload}) @@ -1200,6 +1272,8 @@ async def test_send_command_raises_after_reconnect_retry_also_fails( fake.post = _post await coordinator.async_send_command(bound, 5) + assert coordinator.observe_mode == MODE_POLL + async def test_send_command_reconnect_downgrades_observe_mode( hass: HomeAssistant, From e3e7f4f43c6829a4406823eee3a790fe4b298ebb Mon Sep 17 00:00:00 2001 From: Marc Billow Date: Thu, 6 Aug 2026 02:07:39 +0000 Subject: [PATCH 7/7] test: suppress ty's invalid-assignment on the fake-session swap coordinator is explicitly typed as LocalThingsCoordinator here, so ty correctly sees _session's declared type (DtlsCoapSession | None) and flags assigning a FakeObserveSession to it. The fixture's own _connect_session replacement does the same swap without tripping ty, but only because its self parameter is unannotated -- ty has nothing to check the assignment against there. Deliberate here (this is the whole point of the test: substitute a stand-in session), so silenced rather than restructured; ty's --add-ignore confirmed the comment syntax (ty: ignore[...], not the mypy-style type: ignore[...] used elsewhere in this suite, which ty doesn't appear to honor for this rule). --- tests/localthings/test_coordinator.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index e77da3b..cc289fc 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -772,7 +772,7 @@ async def test_attempt_observe_mode_discards_stale_commit_after_session_swap( other = FakeObserveSession() def _swap_session_mid_wait(subscribed, grace_period_s, success_fraction=None): - coordinator._session = other + coordinator._session = other # ty: ignore[invalid-assignment] return True with patch.object(