diff --git a/AGENTS.md b/AGENTS.md index f096676..4df7a88 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,7 +8,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `pyproject.toml` has a `[tool.mypy]` config but no dev-dependency group declares mypy, so `uv sync` alone won't install it. CI installs it ephemerally via `uv run --with mypy mypy teslemetry_stream`. - `Signal` in `const.py` tracks ; the config route rejects names it does not know with `fst_err_validation`. Fields the API has retired are not rejected - it accepts the request and names them in a top-level `ignoredFields` list - so the library can lag the published list without breaking. - Config responses are shaped inconsistently: success is flat, `{"updated_vehicles": n}` plus `ignoredFields` when some were dropped, while errors are wrapped, `{"response": null, "error": ...}`. Do not look for `updated_vehicles` under `response`; that lookup silently never matches. -- `update_config` debounces changes for 1s into one batch, and the API merges that batch into the vehicle's *full* config before forwarding it to Tesla. `tests/test_config_update.py` covers the response handling. +- `update_config` funnels every caller through one per-vehicle single-flight flush (`TeslemetryStreamVehicle._flush`): the first caller starts it, later callers merge into the same pending config and await it rather than starting their own PATCH. This exists because a batch of listeners scheduled at once (e.g. HA integration setup) must produce one PATCH, not one per listener - see `tests/test_batch_retry_storm.py`. A body-shaped error (`{"error": ...}`) is terminal for that batch: it is not replayed, but the pending config is kept for the next explicit `update_config` call. A transport-level failure (`aiohttp.ClientError`/timeout) gets one bounded retry inside the same flush. `tests/test_config_update.py` covers the response-shape handling. ## Maintaining this file diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 1b22d6b..dbfbfc6 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -6,6 +6,8 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Callable, cast +import aiohttp + from .const import ( BMSState, CabinOverheatProtectionModeState, @@ -67,6 +69,7 @@ class TeslemetryStreamVehicle: fields: dict[str, dict[str, int]] preferTyped: bool | None _config: dict[str, Any] + _flight: asyncio.Task[None] | None def __init__(self, stream: TeslemetryStream, vin: str): # A dictionary of TelemetryField keys and null values @@ -77,6 +80,10 @@ def __init__(self, stream: TeslemetryStream, vin: str): self.fields = {} self.preferTyped = None self._config = {} + # The single in-flight (or most recently completed) coalesced flush. + # Callers that arrive while it is running merge into `_config` and + # await it instead of starting their own PATCH. + self._flight = None @property def config(self) -> dict[str, Any]: @@ -107,11 +114,32 @@ async def get_config(self) -> None: req.raise_for_status() async def update_config(self, config: dict[str, Any]) -> None: - """Update the configuration for the vehicle.""" + """Request a configuration update for the vehicle. + + Merges into the pending desired config and joins the single in-flight + flush for this vehicle rather than starting a new one, so that a + batch of listeners scheduled at the same time produces one PATCH. + """ - # Lock so that we dont change the config while making the API call async with self.lock: self._config = merge(config, self._config) + flight = self._flight + if flight is None or flight.done(): + flight = asyncio.ensure_future(self._flush()) + self._flight = flight + + # Shielded: if this caller is cancelled or times out, it must not + # cancel the shared flush that every other waiter is also awaiting. + # The cancelled caller still sees CancelledError; the flush continues. + await asyncio.shield(flight) + + async def _flush(self) -> None: + """Coalesce the pending config into one PATCH and apply the result. + + On a deterministic upstream rejection (an error-shaped body), the + batch is terminal: it is not replayed, but the desired config is + left in place for the next explicit trigger to attempt again. + """ await asyncio.sleep(1) @@ -119,7 +147,10 @@ async def update_config(self, config: dict[str, Any]) -> None: if not self._config: return - data = await self.patch_config(self._config) + data = await self._patch_with_bounded_retry(self._config) + if data is None: + return + if error := data.get("error"): LOGGER.error( "Error updating streaming config for %s: %s", self.vin, error @@ -151,6 +182,29 @@ async def update_config(self, config: dict[str, Any]) -> None: self.preferTyped = prefer_typed self._config.clear() + async def _patch_with_bounded_retry( + self, config: dict[str, Any] + ) -> dict[str, Any] | None: + """PATCH the config, allowing one retry for a transport-level failure. + + A response that the server actually answered (including an + error-shaped body) is not a transport failure and is returned as-is + for the caller to classify. + """ + for attempt in (1, 2): + try: + return await self.patch_config(config) + except (aiohttp.ClientError, asyncio.TimeoutError) as error: + LOGGER.error( + "Network error updating streaming config for %s (attempt %s/2): %s", + self.vin, + attempt, + error, + ) + if attempt == 1: + await asyncio.sleep(1) + return None + async def patch_config(self, config: dict[str, Any]) -> dict[str, Any]: """Modify the configuration for the vehicle.""" headers = await self.stream.headers() diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py new file mode 100644 index 0000000..116a4d4 --- /dev/null +++ b/tests/test_batch_retry_storm.py @@ -0,0 +1,174 @@ +"""Regression test for the config-write retry storm. + +At integration setup, every streaming listener schedules its own +``add_field`` task. Before the single-flight fix, a deterministic upstream +rejection left the merged pending config in place, so every already-scheduled +listener task replayed the identical PATCH serially - one full write per +listener. This asserts the batch collapses to exactly one upstream call, and +that a later explicit trigger (not the same batch) still gets to retry. + +It also covers a follow-on defect in the single-flight fix itself: a waiter +is joined to the shared flush with a plain ``await``, so cancelling (or +timing out) one caller propagates into the shared task and cancels the +flush out from under every other waiter, leaving the coalesced config +unapplied. Waiters must join the flush shielded from their own cancellation. +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + +ACCEPTED: dict[str, Any] = {"updated_vehicles": 1} +FAILED: dict[str, Any] = {"response": None, "error": "upstream internal error"} + + +class FakeStream: + """Minimal stand-in for TeslemetryStream.""" + + manual = True + + +def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle: + """Build a vehicle that records payloads and replays canned responses.""" + vehicle = TeslemetryStreamVehicle(FakeStream(), vin) # type: ignore[arg-type] + vehicle.sent = [] # type: ignore[attr-defined] + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + vehicle.sent.append(dict(config)) # type: ignore[attr-defined] + return responses.pop(0) if responses else ACCEPTED + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + return vehicle + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<68} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def main() -> None: + results = [] + + # 80 listener tasks race to add distinct fields at setup time, exactly as + # HA's async_setup_stream fans out one add_field task per streaming + # entity. Upstream deterministically rejects every write for this VIN + # (Tesla's per-VIN write latch), so a canned response is queued for each + # possible replay to prove the fix - not just the buggy code's first one. + responses: list[Any] = [FAILED] * 100 + vehicle = make_vehicle(VIN, responses) + fields = [f"Field{i}" for i in range(80)] + await asyncio.wait_for( + asyncio.gather(*(vehicle.add_field(f) for f in fields)), timeout=5 + ) + + results.append( + check( + "a failed batch produces exactly one upstream PATCH", + len(vehicle.sent) == 1, # type: ignore[attr-defined] + f"sent {len(vehicle.sent)} PATCH(es)", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "the single PATCH carries every listener's requested field", + set(vehicle.sent[0]["fields"]) == set(fields), # type: ignore[attr-defined] + f"fields sent {sorted(vehicle.sent[0]['fields'])}", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "no field is recorded as configured after a terminal failure", + vehicle.fields == {}, + f"fields {sorted(vehicle.fields)}", + ) + ) + results.append( + check( + "desired state survives the terminal failure for a later retry", + set(vehicle._config.get("fields", {})) == set(fields), + f"pending {sorted(vehicle._config.get('fields', {}))}", + ) + ) + + # A later explicit trigger (e.g. a new listener) gets its own single + # attempt carrying the still-pending fields, not a fresh replay storm. + # Upstream has since recovered. + responses.clear() + responses.append(ACCEPTED) + await vehicle.add_field("LaterField") + + results.append( + check( + "a later explicit trigger performs exactly one more PATCH", + len(vehicle.sent) == 2, # type: ignore[attr-defined] + f"sent {len(vehicle.sent)} PATCH(es) total", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "the later PATCH carries the still-pending fields plus the new one", + set(vehicle.sent[1]["fields"]) # type: ignore[attr-defined] + == set(fields) | {"LaterField"}, + f"fields sent {sorted(vehicle.sent[1]['fields'])}", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "recovery is recorded once upstream accepts the retry", + all(f in vehicle.fields for f in fields) and "LaterField" in vehicle.fields, + f"fields {sorted(vehicle.fields)}", + ) + ) + + # A batch of concurrent callers joins one shared flush. Cancelling one of + # them mid-flight (here, during the debounce sleep, before the PATCH is + # even sent) must not cancel the flush out from under the others. + vehicle = make_vehicle(VIN, [ACCEPTED]) + cancel_fields = [f"CancelField{i}" for i in range(5)] + tasks = [asyncio.ensure_future(vehicle.add_field(f)) for f in cancel_fields] + await asyncio.sleep(0.05) # let every caller merge in and join the flush + cancelled_task = tasks[2] + cancelled_task.cancel() + + cancelled_raised = False + try: + await cancelled_task + except asyncio.CancelledError: + cancelled_raised = True + + remaining = [t for t in tasks if t is not cancelled_task] + await asyncio.wait_for(asyncio.gather(*remaining), timeout=5) + + results.append( + check( + "the cancelled caller itself observes CancelledError", + cancelled_raised, + ) + ) + results.append( + check( + "cancelling one waiter does not abort the shared flush", + len(vehicle.sent) == 1, # type: ignore[attr-defined] + f"sent {len(vehicle.sent)} PATCH(es)", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "the other concurrent callers still complete with the applied config", + all(f in vehicle.fields for f in cancel_fields), + f"fields {sorted(vehicle.fields)}", + ) + ) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())