Serialize forced coordinator refreshes per entry
NestQuestCoordinator.async_refresh holds a per-entry lock so overlapping service-path refreshes never interleave; the queued pass always publishes after the in-flight one.
This commit is contained in:
@@ -25,6 +25,7 @@ the admin header degrades to a harmless zero instead of guessing.
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import datetime
|
import datetime
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
@@ -97,7 +98,16 @@ class NestQuestSnapshot:
|
|||||||
|
|
||||||
|
|
||||||
class NestQuestCoordinator(DataUpdateCoordinator):
|
class NestQuestCoordinator(DataUpdateCoordinator):
|
||||||
"""Shared refresh cycle for every NestQuest entity."""
|
"""Shared refresh cycle for every NestQuest entity.
|
||||||
|
|
||||||
|
Forced refreshes (the service paths pushing an immediate update
|
||||||
|
after a completion) are SERIALIZED per entry: overlapping
|
||||||
|
complete/uncomplete calls must not interleave their snapshot
|
||||||
|
passes, or an older pass could publish a stale snapshot AFTER a
|
||||||
|
newer one and leave entities wrong until the next poll. Real
|
||||||
|
HA's coordinator already serializes internally; the lock keeps
|
||||||
|
the same guarantee on the stand-in and costs nothing upstream.
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@@ -122,6 +132,14 @@ class NestQuestCoordinator(DataUpdateCoordinator):
|
|||||||
)
|
)
|
||||||
self.entry_id = entry_id
|
self.entry_id = entry_id
|
||||||
self.database = database
|
self.database = database
|
||||||
|
self._refresh_lock = asyncio.Lock()
|
||||||
|
|
||||||
|
async def async_refresh(self) -> None:
|
||||||
|
"""One refresh at a time; a queued refresh runs AFTER the
|
||||||
|
in-flight one publishes, so the last snapshot always reflects
|
||||||
|
the latest mutation."""
|
||||||
|
async with self._refresh_lock:
|
||||||
|
await super().async_refresh()
|
||||||
|
|
||||||
def _local_now(self) -> datetime.datetime:
|
def _local_now(self) -> datetime.datetime:
|
||||||
time_zone = ZoneInfo(self.hass.config.time_zone)
|
time_zone = ZoneInfo(self.hass.config.time_zone)
|
||||||
|
|||||||
@@ -352,3 +352,28 @@ async def test_uncompletion_updates_entities_immediately(
|
|||||||
assert completed_after_uncomplete == 0, (
|
assert completed_after_uncomplete == 0, (
|
||||||
"sensor must reflect the reversal without a manual refresh"
|
"sensor must reflect the reversal without a manual refresh"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_forced_refreshes_serialize_no_stale_publish(
|
||||||
|
hass, make_entry
|
||||||
|
) -> None:
|
||||||
|
"""Overlapping forced refreshes never interleave: a slow pass
|
||||||
|
pauses mid-snapshot, and the queued pass still publishes AFTER
|
||||||
|
it, so the last published snapshot reflects the latest mutation."""
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
entry, child, instances = await _setup_and_seed(hass, make_entry)
|
||||||
|
coordinator = entry.runtime_data.coordinator
|
||||||
|
order: list[str] = []
|
||||||
|
|
||||||
|
async def _slow_update():
|
||||||
|
order.append("start")
|
||||||
|
await asyncio.sleep(0.01)
|
||||||
|
order.append("end")
|
||||||
|
return coordinator.data
|
||||||
|
|
||||||
|
coordinator._async_update_data = _slow_update
|
||||||
|
await asyncio.gather(coordinator.async_refresh(), coordinator.async_refresh())
|
||||||
|
assert order == ["start", "end", "start", "end"], (
|
||||||
|
"overlapping refreshes must serialize: " + str(order)
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user