diff --git a/AGENTS.md b/AGENTS.md index 8930b8a..f718f8f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `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)` and `async_add_connection_listener(callback)`. - `add_field`/`prefer_typed` gate their no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields`/`preferTyped` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. - `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 the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). -- `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. +- `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. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `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, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 2c560b4..021f34f 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -224,8 +224,13 @@ def remove_listener() -> None: def _update_connection_listeners(self, value: bool | None = None) -> None: """Update all connection listeners with retry count""" - for listener in self._connection_listeners.values(): - listener(self.connected if value is None else value) + # A snapshot, not a live view - a callback that creates a vehicle + # (get_vehicle) or otherwise adds a connection listener mid-dispatch + # must not mutate _connection_listeners while this is iterating it, + # which would raise RuntimeError and kill connect()/disconnect. + connected = self.connected if value is None else value + for listener in list(self._connection_listeners.values()): + listener(connected) async def connect(self) -> None: """ diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 146f898..8c2b748 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -512,6 +512,51 @@ async def get_config() -> None: await asyncio.sleep(0) +async def test_connect_survives_connection_listener_creating_vehicle_mid_dispatch( + results: list[bool], +) -> None: + """A connection listener that calls get_vehicle() for an uncached VIN - + e.g. an integration reacting to reconnect by discovering a vehicle - + registers a new connection listener (TeslemetryStreamVehicle.__init__) + while _update_connection_listeners() is iterating _connection_listeners. + That must not raise and kill connect().""" + session = FakeSession() + stream = make_stream(session, manual=True) + + discovered: list[TeslemetryStreamVehicle] = [] + + def on_connect(connected: bool) -> None: + if connected and not discovered: + discovered.append(stream.get_vehicle("NEWVIN1")) + + stream.async_add_connection_listener(on_connect) + + try: + await stream.connect() + connect_ok = True + except RuntimeError as error: + connect_ok = False + connect_error = repr(error) + results.append( + check( + "connect() survives a connection listener mutating _connection_listeners", + connect_ok, + "" if connect_ok else connect_error, + ) + ) + results.append(check("the listener discovered the new vehicle", len(discovered) == 1)) + results.append( + check( + "the new vehicle's own connection listener is registered", + len(stream._connection_listeners) == 2, + f"connection listeners {len(stream._connection_listeners)}", + ) + ) + + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -523,6 +568,7 @@ async def main() -> None: await test_dispatch_survives_listener_creating_vehicle_mid_iteration(results) await test_internal_listener_sees_event_before_public_mutator(results) await test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use(results) + await test_connect_survives_connection_listener_creating_vehicle_mid_dispatch(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT")