From 5ef5a76a191e2bd45b9e2b6a8e74c3ada335c765 Mon Sep 17 00:00:00 2001 From: Brett Date: Mon, 27 Jul 2026 09:58:10 +1000 Subject: [PATCH 1/2] fix(vehicle): single-flight config writes to stop the batch replay storm update_config() logged and returned on an error-shaped response without clearing the merged pending config, so every already-scheduled listener task (HA's async_setup_stream fans out one add_field task per streaming entity) independently replayed the identical full PATCH - 80 requested fields observed replaying serially against a deterministic upstream rejection. TeslemetryStreamVehicle now funnels every caller through one per-vehicle flush: the first caller starts it, later callers merge into the same pending config and await that flush instead of starting their own. A body-shaped error is terminal for the batch - it is not replayed, but the desired config is kept for the next explicit update_config call (a new listener, reconnect, or manual retry). A transport-level failure gets one bounded retry inside the same flush before it is treated as failed. --- AGENTS.md | 2 +- teslemetry_stream/vehicle.py | 57 +++++++++++++- tests/test_batch_retry_storm.py | 128 ++++++++++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 4 deletions(-) create mode 100644 tests/test_batch_retry_storm.py 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..3f52958 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,29 @@ 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 + + await 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 +144,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 +179,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..9694eae --- /dev/null +++ b/tests/test_batch_retry_storm.py @@ -0,0 +1,128 @@ +"""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. +""" +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)}", + ) + ) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) From 4919602b585670946adeb5a127d34e92c22190f6 Mon Sep 17 00:00:00 2001 From: Brett Date: Mon, 27 Jul 2026 10:07:02 +1000 Subject: [PATCH 2/2] fix(vehicle): shield single-flight waiters from their own cancellation A waiter joined the shared per-vehicle flush with a plain await, so cancelling (or timing out) one caller threw CancelledError into the flush itself, aborting it for every other concurrent waiter and leaving the coalesced config unapplied. Join with asyncio.shield instead: the cancelled caller still observes CancelledError, but the flush it was waiting on keeps running to completion for the remaining waiters. --- teslemetry_stream/vehicle.py | 5 +++- tests/test_batch_retry_storm.py | 46 +++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 3f52958..dbfbfc6 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -128,7 +128,10 @@ async def update_config(self, config: dict[str, Any]) -> None: flight = asyncio.ensure_future(self._flush()) self._flight = flight - await 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. diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index 9694eae..116a4d4 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -6,6 +6,12 @@ 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 @@ -118,6 +124,46 @@ async def main() -> None: ) ) + # 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):