diff --git a/AGENTS.md b/AGENTS.md index 54f3c39..988c6a8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,8 +18,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. - `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `__init__` registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, unconditionally at construction - not lazily - so no connection can ever predate it and miss an event. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. The stored `fields` dict (and each nested per-field dict) is copied, never the same object handed to public listeners for that same event - a consumer mutating its event in place must not corrupt the record. - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)`. -- The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. -- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. +- The config-sync listener can only observe a server-side change while connected, while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`, AND after this connection's own config snapshot has actually been applied. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (all three true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. The third condition exists because `TeslemetryStream.connected` flips true as soon as `connect()` installs the response, before `listen()` has dispatched a single event - a connection listener reacting to reconnect can otherwise run `add_field`/`prefer_typed` in that window and match against the stale pre-disconnect record. `TeslemetryStream._connection_id` (bumped once per successful `connect()`) and `TeslemetryStreamVehicle._synced_connection_id` (set to it in `_on_config_event`) close that gap: `_record_is_live()` also requires the two to match. Both default to `0`/read via `getattr(..., 0)` so stream stand-ins that skip `_connection_id` entirely (most test doubles, which never call the real `connect()`) keep the pre-existing "live" behavior. +- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case; `tests/test_reconnect_config_window.py` covers the reconnect-window race (connected-but-presnapshot no-op skip). - `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, and internal-before-public dispatch order. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 2c560b4..9fe7c17 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -20,6 +20,10 @@ class TeslemetryStream: """Teslemetry Stream Client""" _response: aiohttp.ClientResponse | None = None + # Bumped each time connect() installs a new response - lets a vehicle + # tell "this connection's config snapshot has been applied" apart from + # "some past connection's was". + _connection_id: int = 0 def __init__( self, @@ -267,6 +271,7 @@ async def connect(self) -> None: if self._response is not None: self._response.close() self._response = response + self._connection_id += 1 LOGGER.debug( "Connected to %s with status %s", self._response.url, self._response.status ) diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index e28c5f6..fdb8413 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -83,6 +83,11 @@ def __init__(self, stream: TeslemetryStream, vin: str): self.fields = {} self.preferTyped = None self._config = {} + # The connection id (see TeslemetryStream._connection_id) whose + # snapshot this record last reflects - gates _record_is_live so a + # reconnect can't be mistaken for "already synced" before its own + # config event has actually arrived. + self._synced_connection_id = getattr(stream, "_connection_id", 0) # 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. @@ -140,6 +145,8 @@ def _on_config_event(self, event: dict[str, Any]) -> None: ) return + self._synced_connection_id = getattr(self.stream, "_connection_id", 0) + if "fields" in config: fields = config["fields"] # Every entry must itself be a dict (e.g. {"interval_seconds": 60} @@ -317,13 +324,21 @@ def _record_is_live(self) -> bool: The skip is purely an optimization - the server handles a redundant PATCH fine - so this only needs to answer "is the config-sync listener actually able to observe a server-side change right now", - not force the record fresh. That requires both a live connection and - the `config` topic not being filtered out via `TeslemetryStream - (topics=...)`; if either is false, add_field/prefer_typed skip the - no-op check and always send, same as the pre-feature status quo. + not force the record fresh. That requires a live connection, the + `config` topic not being filtered out via `TeslemetryStream + (topics=...)`, AND this connection's own config snapshot having + already been applied - `connected` flips true as soon as the + response is installed, before that snapshot has been dispatched, so + without this last check a listener reacting to reconnect could match + against the pre-disconnect record and skip a PATCH that server-side + drift actually required. If any of the three is false, + add_field/prefer_typed skip the no-op check and always send, same as + the pre-feature status quo. """ if not self.stream.connected: return False + if getattr(self.stream, "_connection_id", 0) != self._synced_connection_id: + return False topics = self.stream.topics return topics is None or SseTopic.CONFIG in topics diff --git a/tests/test_reconnect_config_window.py b/tests/test_reconnect_config_window.py new file mode 100644 index 0000000..3645d0c --- /dev/null +++ b/tests/test_reconnect_config_window.py @@ -0,0 +1,138 @@ +"""Regression test for a reconnect-window gap in the config no-op skip. + +``TeslemetryStream.connect()`` sets ``self._response`` and notifies +connection listeners (``stream.connected`` becomes True) before ``listen()`` +has read a single SSE line - in particular, before the server's post-connect +config snapshot has been dispatched to the vehicle's internal config +listener. ``add_field``/``prefer_typed``'s no-op skip is gated on +``_record_is_live()``, which only checks "connected and not topic-filtered" - +not "has this connection's snapshot actually been applied yet". A connection +listener that calls ``add_field()`` in that window can match against the +stale pre-disconnect record and skip its PATCH, and nothing retries it once +the snapshot later reveals the mismatch. +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from teslemetry_stream.const import Key +from teslemetry_stream.stream import TeslemetryStream +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + + +class FakeResponse: + """A response whose content is never actually iterated in this test.""" + + status = 200 + url = "https://api.teslemetry.com/sse" + + def close(self) -> None: + pass + + +class FakeSession: + """A session whose get() hands back a connected-but-empty response.""" + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + return FakeResponse() + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +def make_stream(**kwargs: Any) -> TeslemetryStream: + kwargs.setdefault("manual", True) + return TeslemetryStream( + session=FakeSession(), # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + **kwargs, + ) + + +def make_vehicle_with_capture( + stream: TeslemetryStream, +) -> tuple[TeslemetryStreamVehicle, list[dict[str, Any]]]: + vehicle = TeslemetryStreamVehicle(stream, VIN) + sent: list[dict[str, Any]] = [] + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + sent.append(dict(config)) + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + return vehicle, sent + + +async def test_reconnect_window_add_field_matches_stale_record( + results: list[bool], +) -> None: + stream = make_stream() + vehicle, sent = make_vehicle_with_capture(stream) + + # Pre-disconnect state: server had BatteryLevel @ 60s. + vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} + results.append(check("stream starts disconnected", not stream.connected)) + + # A connection listener reacting to reconnect - e.g. an HA integration + # re-asserting its desired fields - the exact scenario `_record_is_live` + # exists to protect against a stale match. + scheduled: list[asyncio.Task[None]] = [] + + def on_connect(connected: bool) -> None: + if connected: + scheduled.append(asyncio.ensure_future(vehicle.add_field("BatteryLevel", 60))) + + stream.async_add_connection_listener(on_connect) + + # Reconnect: this flips `connected` True and fires `on_connect` + # synchronously, scheduling (not yet running) the add_field task. + await stream.connect() + results.append(check("stream reports connected immediately after connect()", stream.connected)) + + # Let the scheduled add_field task run its no-op check BEFORE the + # server's config snapshot for this connection has arrived - the window + # under test. Server-side truth (unbeknownst to this client yet) is + # actually 30s, not the stale 60s the record still holds. + await asyncio.sleep(0) + + # Now the snapshot for the new connection arrives. + vehicle._on_config_event( + {Key.VIN: VIN, Key.CONFIG: {"fields": {"BatteryLevel": {"interval_seconds": 30}}}} + ) + + await asyncio.gather(*scheduled) + + results.append( + check( + "add_field sends a PATCH instead of matching the pre-reconnect stale record", + len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + f"sent {sent}", + ) + ) + results.append( + check( + "the desired 60s interval is the vehicle's final state, not the stale 30s snapshot", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, + f"fields {vehicle.fields}", + ) + ) + + +async def main() -> None: + results: list[bool] = [] + await test_reconnect_window_add_field_matches_stale_record(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())