diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index 5682454..3a92961 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -91,7 +91,6 @@ _SEED_PATH = ["device", "0"] # config entry -- it's device state, not configuration, and runs to tens of # kilobytes. _SNAPSHOT_VERSION = 1 -_SNAPSHOT_SAVE_DELAY_S = 10.0 def snapshot_store(hass: HomeAssistant, entry: ConfigEntry) -> Store[dict[str, Any]]: @@ -1060,8 +1059,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): agree here would register byte-identical entities.""" return frozenset((b.subdevice.key, _key(b)) for b in self.bound) - @callback - def _save_snapshot(self, resources: dict[str, dict], candidates: list[Subdevice]) -> None: + async def _async_save_snapshot( + self, resources: dict[str, dict], candidates: list[Subdevice] + ) -> None: """Record what this first cycle handed `_run_discovery`, so the next restart can replay it while the appliance is unreachable. @@ -1069,14 +1069,29 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): `discover_partitioned` takes candidates and returns the live ones, so replaying against the narrowed list would rediscover nothing for a composite appliance's siblings. + + Written now rather than through `async_delay_save`, because a pending + delayed write outlives whatever queued it: it lands after + `async_remove_entry` has deleted the file and recreates it orphaned, + and a reload scheduled by `_reconcile_rehydrated` would read the + pre-reload snapshot back off disk. This runs once per entry load, so + the immediate write costs nothing worth deferring. """ ident = self._identity - payload = { - "resources": dict(resources), - "subdevice_candidates": [asdict(su) for su in candidates], - "identity": asdict(ident) if ident is not None else None, - } - self._snapshot_store.async_delay_save(lambda: payload, _SNAPSHOT_SAVE_DELAY_S) + try: + await self._snapshot_store.async_save( + { + "resources": dict(resources), + "subdevice_candidates": [asdict(su) for su in candidates], + "identity": asdict(ident) if ident is not None else None, + } + ) + except Exception as e: + # Never fail a poll over the snapshot -- a board reporting + # something the JSON encoder rejects would otherwise break + # polling outright. Worst case this entry can't load offline, + # which is where it was before any of this existed. + self._log.warning("could not write discovery snapshot: %s", e) async def async_rehydrate(self) -> bool: """Register the last known entity set without reaching the device. @@ -1098,37 +1113,40 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): if not stored or not stored.get("resources"): return False - ident = stored.get("identity") - if ident is not None: - self._identity = DeviceIdentity( - manufacturer=ident.get("manufacturer") or "", - model=ident.get("model") or "", - name=ident.get("name") or "", - serial=ident.get("serial"), - device_types=tuple(ident.get("device_types") or ()), - raw=ident.get("raw") or {}, - ) - # JSON gives lists back where Subdevice declares tuples, and it's a - # frozen (hashable) dataclass used as a dict key in flatten(). - self.subdevices = [ - Subdevice( - kind=su["kind"], - key=su["key"], - seed_path=tuple(su.get("seed_path") or ()), - flat_hrefs=tuple(su.get("flat_hrefs") or ()), - ) - for su in stored.get("subdevice_candidates") or () - ] - resources = stored["resources"] try: + ident = stored.get("identity") + if ident is not None: + self._identity = DeviceIdentity( + manufacturer=ident.get("manufacturer") or "", + model=ident.get("model") or "", + name=ident.get("name") or "", + serial=ident.get("serial"), + device_types=tuple(ident.get("device_types") or ()), + raw=ident.get("raw") or {}, + ) + # JSON gives lists back where Subdevice declares tuples, and it's + # a frozen (hashable) dataclass used as a dict key in flatten(). + self.subdevices = [ + Subdevice( + kind=su["kind"], + key=su["key"], + seed_path=tuple(su.get("seed_path") or ()), + flat_hrefs=tuple(su.get("flat_hrefs") or ()), + ) + for su in stored.get("subdevice_candidates") or () + ] self._run_discovery(resources, from_snapshot=True) except Exception as e: - # A snapshot written by an older release can outlive the registry - # shape it was discovered against. Fall back to the pre-#295 - # behavior rather than failing the entry outright. + # A snapshot written by an older release can outlive both the + # stored shape and the registry it was discovered against. This + # has to catch the rebuild as well as the replay: an exception + # escaping here reaches async_setup_entry, which only handles + # ConfigEntryNotReady, so the entry would land in SETUP_ERROR -- + # never retried, and with its session left open. self._log.warning("discovery snapshot could not be replayed: %s", e, exc_info=True) self.bound = [] + self.subdevices = [] return False # _run_discovery sets _discovered; put it back. The snapshot only @@ -1603,7 +1621,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): # the passed dict, never the cache. candidates = list(self.subdevices) self._run_discovery(resources) - self._save_snapshot(resources, candidates) + # Banked before the reconcile below, so a reload it schedules + # comes up against this discovery rather than the one that is + # being replaced. + await self._async_save_snapshot(resources, candidates) self._reconcile_rehydrated() resources = self._live_subdevice_resources(resources) sweep_mismatch = False diff --git a/docs/investigations/offline-setup.md b/docs/investigations/offline-setup.md index 95b812e..7ff350f 100644 --- a/docs/investigations/offline-setup.md +++ b/docs/investigations/offline-setup.md @@ -93,6 +93,14 @@ It lives in `.storage` (`Store`, keyed on entry_id) rather than on the config entry: it's device state, not configuration, and runs to tens of kilobytes. `async_remove_entry` deletes it with the entry. +The write is awaited, not `async_delay_save`d. A deferred write outlives +whatever queued it: it lands after `async_remove_entry` has deleted the file +and recreates it orphaned, and a reload scheduled by the reconcile below +would read the pre-reload snapshot back off disk. It runs once per entry +load, so there's nothing worth deferring. A write that fails is logged and +swallowed — a board reporting something the JSON encoder rejects must not +break polling. + ### 2. Reconcile on reconnect The snapshot is a claim about a device we haven't talked to yet. When the diff --git a/tests/localthings/test_offline_setup.py b/tests/localthings/test_offline_setup.py index 067038f..dc9597f 100644 --- a/tests/localthings/test_offline_setup.py +++ b/tests/localthings/test_offline_setup.py @@ -23,10 +23,7 @@ from homeassistant.util import dt as dt_util from pytest_homeassistant_custom_component.common import async_fire_time_changed from custom_components.localthings.const import DOMAIN, SUMMARY_INTERVAL_S -from custom_components.localthings.coordinator import ( - _SNAPSHOT_SAVE_DELAY_S, - LocalThingsCoordinator, -) +from custom_components.localthings.coordinator import LocalThingsCoordinator from custom_components.localthings.registry.identity import DeviceIdentity from .conftest import _load_fridge_resources as _load_fridge @@ -65,12 +62,6 @@ def _store_key(entry) -> str: return f"{DOMAIN}.{entry.entry_id}.discovery" -async def _flush_snapshot(hass: HomeAssistant) -> None: - """Run out `async_delay_save`'s timer without reaching the next poll.""" - async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=_SNAPSHOT_SAVE_DELAY_S + 1)) - await hass.async_block_till_done() - - async def _tick(hass: HomeAssistant) -> None: """Advance past one summary interval so the coordinator polls again. @@ -89,7 +80,6 @@ async def _setup_online_then_unload(hass: HomeAssistant, entry, resources: dict) await hass.config_entries.async_setup(entry.entry_id) await hass.async_block_till_done() entity_ids = {s.entity_id for s in hass.states.async_all()} - await _flush_snapshot(hass) await hass.config_entries.async_unload(entry.entry_id) await hass.async_block_till_done() return entity_ids @@ -106,7 +96,6 @@ async def test_snapshot_written_after_first_discovery( """A successful first cycle banks what it handed _run_discovery.""" await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() - await _flush_snapshot(hass) stored = hass_storage[_store_key(mock_entry)]["data"] assert stored["resources"] @@ -122,11 +111,7 @@ async def test_snapshot_not_written_when_device_never_answers( with _unreachable(): await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() - # Asserted before the flush: running the save timer out also runs out - # HA's own setup-retry timer, which puts the entry back into - # SETUP_IN_PROGRESS. assert mock_entry.state is ConfigEntryState.SETUP_RETRY - await _flush_snapshot(hass) assert _store_key(mock_entry) not in hass_storage @@ -135,14 +120,22 @@ async def test_snapshot_removed_when_entry_removed( hass: HomeAssistant, mock_entry, mock_coordinator_session, hass_storage ) -> None: """The store is keyed on entry_id, so re-adding the appliance mints a new - one -- the old file has to go with the entry that wrote it.""" + one -- the old file has to go with the entry that wrote it. + + The clock is run on afterwards because a deferred write would land here: + with `async_delay_save` the removal was undone a few seconds later by the + save the last poll had queued, leaving the file orphaned for good. + """ await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() - await _flush_snapshot(hass) assert _store_key(mock_entry) in hass_storage await hass.config_entries.async_remove(mock_entry.entry_id) await hass.async_block_till_done() + assert hass_storage.get(_store_key(mock_entry), {}).get("data") is None + + async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=60)) + await hass.async_block_till_done(wait_background_tasks=True) assert hass_storage.get(_store_key(mock_entry), {}).get("data") is None @@ -200,6 +193,39 @@ async def test_offline_load_without_snapshot_still_fails(hass: HomeAssistant, mo assert not hass.states.async_all() +async def test_malformed_snapshot_falls_back_to_setup_retry( + hass: HomeAssistant, mock_entry, hass_storage +) -> None: + """A stored row missing a field the current dataclass declares must fail + the same way an unreachable device does. + + Anything escaping async_rehydrate reaches async_setup_entry, which only + handles ConfigEntryNotReady -- so the entry would land in SETUP_ERROR, + which HA never retries, with its DTLS session left open on the fixed + source port the next attempt binds. + """ + key = _store_key(mock_entry) + hass_storage[key] = { + "version": 1, + "minor_version": 1, + "key": key, + "data": { + "resources": _load_fridge(), + "subdevice_candidates": [{"key": "1"}], # no "kind" + }, + } + + with ( + _unreachable(), + patch.object(LocalThingsCoordinator, "async_close", autospec=True) as close, + ): + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + assert mock_entry.state is ConfigEntryState.SETUP_RETRY + close.assert_awaited_once() + + async def test_corrupt_snapshot_falls_back_to_setup_retry( hass: HomeAssistant, mock_entry, hass_storage ) -> None: @@ -236,7 +262,6 @@ async def test_snapshot_restores_identity(hass: HomeAssistant, mock_entry) -> No with _reachable(resources, identity): await hass.config_entries.async_setup(mock_entry.entry_id) await hass.async_block_till_done() - await _flush_snapshot(hass) await hass.config_entries.async_unload(mock_entry.entry_id) await hass.async_block_till_done() @@ -278,7 +303,7 @@ async def test_offline_load_keeps_polling_and_recovers(hass: HomeAssistant, mock async def test_reconcile_reloads_when_live_discovery_differs( - hass: HomeAssistant, mock_entry + hass: HomeAssistant, mock_entry, hass_storage ) -> None: """Platforms enumerate `bound` once, so a live set that disagrees with the snapshot can only be adopted by bringing the entry back up.""" @@ -289,7 +314,6 @@ async def test_reconcile_reloads_when_live_discovery_differs( await hass.async_block_till_done() coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] victim = coordinator.bound[0].href - await _flush_snapshot(hass) await hass.config_entries.async_unload(mock_entry.entry_id) await hass.async_block_till_done() @@ -305,6 +329,11 @@ async def test_reconcile_reloads_when_live_discovery_differs( await _tick(hass) reload.assert_called_once_with(mock_entry.entry_id) + # Banked before the reload is scheduled, so the entry that comes back up + # replays this discovery rather than the one it is replacing -- otherwise + # a device that goes quiet again mid-reload rehydrates the stale set and + # reconciles all over again. + assert victim not in hass_storage[_store_key(mock_entry)]["data"]["resources"] async def test_reconcile_is_quiet_when_live_discovery_agrees(