From ae7f68f6da46927387ceef1e446255ba5841d34d Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 15:21:22 +1000 Subject: [PATCH 1/2] fix: give listeners added mid-dispatch a pass at the triggering event A generic listener that discovers an uncached VIN via get_vehicle() while handling that VIN's config event registers a new internal listener after listen() has already snapshotted the registry for the current event. The new vehicle's config-sync listener then misses the very event that revealed its config, leaving it at fields={}/ preferTyped=None until (if ever) a later config event arrives. --- teslemetry_stream/stream.py | 13 ++++++++ tests/test_stream_lifecycle.py | 54 ++++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 2c560b4..63deb10 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -443,6 +443,7 @@ async def listen(self) -> None: # the loop. Internal (bookkeeping) listeners go first, so # one can cache from the pristine event before any public # callback gets a chance to mutate it in place. + dispatched_keys = set(self._listeners.keys()) ordered = sorted(self._listeners.values(), key=lambda item: not item[2]) for listener, filters, _internal in ordered: if recursive_match(filters, event): @@ -450,6 +451,18 @@ async def listen(self) -> None: listener(event) except Exception as error: LOGGER.error("Uncaught error in listener: %s", error) + # A callback above may have added a listener (e.g. + # get_vehicle() for an uncached VIN registers that + # vehicle's internal config listener) that missed the + # snapshot - give listeners added mid-dispatch one more + # pass at this same event so a newly-created vehicle + # isn't seeded stale. + for key, (listener, filters, _internal) in list(self._listeners.items()): + if key not in dispatched_keys and recursive_match(filters, event): + try: + listener(event) + except Exception as error: + LOGGER.error("Uncaught error in listener: %s", error) finally: self._close_response() if self._listen_task is current_task: diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index c99875d..3582c78 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -434,6 +434,59 @@ def public_mutator(event: dict[str, Any]) -> None: await asyncio.sleep(0) +async def test_vehicle_discovered_mid_dispatch_seeds_from_current_config_event( + results: list[bool], +) -> None: + """A generic public listener that discovers an uncached VIN and calls + get_vehicle() while handling that VIN's own config event registers a new + internal listener that the current dispatch already snapshotted past. + Without a second pass over listeners added mid-dispatch, the newly + created vehicle would miss the very event that revealed its config and + sit at fields={}/preferTyped=None until (if ever) another config event + arrives.""" + new_vin = "TESTVIN0000000002" + session = FakeSession() + session.content_factory = lambda: FakeEventContent( + [ + ( + b'data: {"vin": "' + + new_vin.encode() + + b'", "config": {"fields": ' + + b'{"BatteryLevel": {"interval_seconds": 60}}, ' + + b'"prefer_typed": true}}\n' + ) + ] + ) + stream = make_stream(session) + + discovered: list[TeslemetryStreamVehicle] = [] + + def generic_listener(event: dict[str, Any]) -> None: + vin = event.get("vin") + if vin and vin not in stream.vehicles: + discovered.append(stream.get_vehicle(vin)) + + stream.async_add_listener(generic_listener, {"vin": None}) + + for _ in range(5): + await asyncio.sleep(0) + + results.append(check("the listener discovered the new vehicle", len(discovered) == 1)) + if discovered: + vehicle = discovered[0] + results.append( + check( + "the newly discovered vehicle is seeded from the event that created it", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60} + and vehicle.preferTyped is True, + f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}", + ) + ) + + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -444,6 +497,7 @@ async def main() -> None: await test_restart_after_public_readd_with_internal_listener_present(results) 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_seeds_from_current_config_event(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") From f1d43e1b4f97569078b0b30cf2bd8c20d2e1be8e Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 15:21:44 +1000 Subject: [PATCH 2/2] docs: note the mid-dispatch listener second-pass in listen() --- AGENTS.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 54f3c39..3f1e9c2 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)`. - 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. -- `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. A listener added mid-dispatch (not in the original snapshot) still gets one extra pass at the *same* event once the snapshot's dispatch finishes - otherwise a vehicle created by `get_vehicle()` from within that event's dispatch would miss the very event that revealed its config and stay seeded empty. `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 mid-dispatch-created vehicle being seeded from its triggering event. ## Maintaining this file