Harden the snapshot path against three review findings

Widen async_rehydrate's guard to cover the identity and Subdevice rebuild,
not just the replay. A stored row missing a field the current dataclass
declares raised KeyError straight out of async_setup_entry, which only
handles ConfigEntryNotReady -- so the entry landed in SETUP_ERROR, which HA
never retries, with its DTLS session left open on the fixed source port the
next attempt binds. It now fails the same way an unreachable device does.

Write the snapshot immediately instead of through async_delay_save. A
deferred write outlives whatever queued it: removing an entry inside the
delay window deleted the file and then had it recreated, orphaned, when the
timer fired; and a reload scheduled by _reconcile_rehydrated read the
pre-reload snapshot back off disk, so a device going quiet again mid-reload
rehydrated the stale set and reconciled a second time. Banking it before the
reconcile fixes the ordering. Failures are logged rather than raised -- a
board reporting something the JSON encoder rejects must not break polling.

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

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

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(