fix(vehicle): single-flight config writes to stop the batch replay storm

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 <https://api.teslemetry.com/fields.json>; 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
 

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,19 +114,43 @@ 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)
 
         async with self.lock:
             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()

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())