From 8b939994387b6f301992a738b966805462f6be34 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 15 Sep 2026 08:30:04 +1000 Subject: [PATCH 1/4] fix: reconnect a silently stalled stream, log a listen task that never connects Add a silence watchdog in __anext__: each SSE read is bounded by a single 30s window (three missed 10s keepalives), so a connection that stops producing any bytes - not just a hard error - gets closed and picked up by the existing reconnect loop. connect()'s sock_read is left unbounded so the two timeouts don't race. listen() now tracks whether it has ever reached a successful connect; if it returns or raises before that, it logs at ERROR and re-raises rather than dying quietly. Motivated by a customer's Home Assistant instance whose listen task never logged a single connect attempt for 2.5+ days after a restart, with no server-side or client-side trace of why. --- AGENTS.md | 5 + teslemetry_stream/stream.py | 59 +++++++- tests/test_listen_early_exit.py | 178 ++++++++++++++++++++++ tests/test_silence_watchdog.py | 251 ++++++++++++++++++++++++++++++++ 4 files changed, 487 insertions(+), 6 deletions(-) create mode 100644 tests/test_listen_early_exit.py create mode 100644 tests/test_silence_watchdog.py diff --git a/AGENTS.md b/AGENTS.md index fdd3d2d..c14c94e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,6 +24,8 @@ Project-intrinsic knowledge that should travel with the code: build, test, relea - Internal reconnect paths (EOF, `ClientError`, unexpected exceptions) call `_close_response()`, **never** `close()` - `close()` additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally so cancellation still releases the connection. - Dispatch iterates a **sorted snapshot** of `_listeners.values()` with internal listeners first, never the live dict: a callback adding a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration`, and a public callback must not mutate the event before an internal bookkeeping listener has cached from it. `_update_connection_listeners()` has the same mutation hazard and iterates a plain `list(...)` snapshot (no ordering requirement - connection listeners share no mutable event). - `__anext__` treats `ClientResponseError` with status 401/403 as terminal: sets `active = False` and raises `TeslemetryStreamAuthenticationError`. Every other `aiohttp.ClientError` keeps backoff-and-reconnect. +- `__anext__` reads each SSE line under `asyncio.wait_for(..., timeout=_SILENCE_WINDOW)` - this is the one mechanism bounding a connection gone quiet (no data, no keepalive) without erroring; `connect()`'s `sock_read` is deliberately unbounded so the two don't race. A timeout closes the response and falls through to the outer loop's immediate reconnect, same as a clean server-side end. `_silence_streak` gates logging: the first silence-triggered reconnect after a real read logs WARNING, further ones with still no read in between log DEBUG - it resets only on a successfully read line (data or keepalive), not on a bare reconnect. +- `listen()` tracks whether it has ever seen a successful connect (via a scoped internal connection listener). If it returns or raises before that - including cancellation - it logs ERROR before propagating; nothing before a first connect is swallowed. - `async_add_listener(..., internal=True)` marks a bookkeeping-only listener (the vehicle config-sync listener is the only user). Its `schedule_refresh` gate is unconditionally false, which is what makes construction-time registration safe outside a running event loop. Both the "first listener starts the task" and "last listener removed auto-closes" checks count only public listeners, so an internal listener can neither pin the connection open nor block a later public listener's zero-to-one restart. **Any stream stand-in (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`.** - `topics=` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param. `SseTopic` is the closed server-recognized set; `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets, flat per-product-kind lists deliberately not split by whether a topic has a connect-time snapshot (server behavior, not something this library encodes). Omitting it (`None`) is legacy-all forever. An explicitly empty iterable raises `ValueError` at construction - "no topics" must not mean "all topics", mirroring the server's 400. A bare `str`/`SseTopic` is accepted as a single topic, not iterated character-by-character; `topics: str | Iterable[str] | None` exists precisely because a lone string also satisfies `Iterable[str]`. - `ingest()` (and `TeslemetryStreamVehicle.ingest()`, same call with the VIN filled in) is the entry point for an observation not read off the SSE connection - a Bluetooth broadcast, today. It builds the native wire event (`vin`/`data`/`createdAt`) plus an open-ended `metadata` dict (`Metadata.SOURCE`/`Metadata.RAW`) and hands it to the same `_dispatch` `listen()` uses, so existing `listen_*` callbacks receive both sources with no translation and no second subscription. Dispatch is arrival-ordered and the stream holds no per-field value: **nothing is deduplicated, reordered, or dropped, and there is deliberately no source ranking or precedence** - which report to believe is the consumer's decision, made on `metadata`. Ingesting neither requires nor opens a connection. Neither library depends on the other: the BLE-side shim shaping a broadcast into this format lives in `tesla-fleet-api` and the consumer wires the two. @@ -55,6 +57,9 @@ Project-intrinsic knowledge that should travel with the code: build, test, relea | Reconnect-window re-fetch race | `test_reconnect_config_window.py` | | Listen task ownership, dispatch snapshots/order, close races | `test_stream_lifecycle.py` | | 401/403 terminal vs transient retry | `test_auth_failure.py` | +| Clean vs abrupt stream end logging | `test_clean_stream_end.py` | +| Silence watchdog: timeout reconnect, keepalive reset, WARNING-once/DEBUG-repeat | `test_silence_watchdog.py` | +| `listen()` early exit before first connect logs ERROR and re-raises | `test_listen_early_exit.py` | | `str` payload coercion against real telemetry | `test_field_type_coercion.py` | | `topics` URL construction, empty/bare-string handling, tariff listener | `test_sse_topics.py` | | Native-event regression, source indistinguishability, no-dedup contract | `test_external_ingest.py` | diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 9bd0ffb..e7835f9 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -16,6 +16,11 @@ LOGGER = logging.getLogger(__package__) +# The server sends an SSE keepalive comment every 10s. Three missed +# keepalives (30s of total silence) bounds reconnecting a connection that +# has gone quiet without erroring. +_SILENCE_WINDOW = 30.0 + class TeslemetryStream: """Teslemetry Stream Client""" @@ -72,6 +77,9 @@ def __init__( ] = {} self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {} self._listen_task: asyncio.Task[None] | None = None + # Consecutive silence-triggered reconnects with no successful read in + # between - gates the WARNING-once/DEBUG-after-that logging split. + self._silence_streak: int = 0 # Created lazily in connect() - asyncio.Lock() requires a running # loop on Python 3.9, and streams are commonly built before one. self._connect_lock: asyncio.Lock | None = None @@ -260,8 +268,11 @@ async def connect(self) -> None: headers=headers, params=params, raise_for_status=True, + # sock_read is intentionally unbounded: the silence watchdog + # in __anext__ is the one mechanism that bounds dead air on + # an open connection, with logging this bare timeout can't do. timeout=aiohttp.ClientTimeout( - connect=5, sock_connect=5, sock_read=30, total=None + connect=5, sock_connect=5, sock_read=None, total=None ), chunked=True, ) @@ -334,14 +345,33 @@ async def __anext__(self) -> dict[str, Any]: # Connect to the stream await self.connect() assert self._response - async for line_in_bytes in self._response.content: + content_iter = self._response.content.__aiter__() + while True: + try: + line_in_bytes = await asyncio.wait_for( + content_iter.__anext__(), timeout=_SILENCE_WINDOW + ) + except StopAsyncIteration: + raise TeslemetryStreamEnded() from None + except asyncio.TimeoutError: + if self._silence_streak == 0: + LOGGER.warning( + "No data received for %ss, reconnecting", _SILENCE_WINDOW + ) + else: + LOGGER.debug( + "No data received for %ss, reconnecting", _SILENCE_WINDOW + ) + self._silence_streak += 1 + self._close_response() + break + self._silence_streak = 0 field, _, value = line_in_bytes.decode("utf8").partition(": ") if field == "data": data = json.loads(value) if self.parse_timestamp: data["timestamp"] = _parse_created_at(data["createdAt"]) return cast(dict[str, Any], data) - raise TeslemetryStreamEnded() except StopAsyncIteration as e: # Re-raise explicitly so it isn't caught by the generic Exception handler below self.disconnect() @@ -446,11 +476,28 @@ async def listen(self) -> None: return self._listen_task = current_task + connected_once = False + + def _mark_connected(value: bool) -> None: + nonlocal connected_once + connected_once = connected_once or value + + remove_connection_listener = self.async_add_connection_listener(_mark_connected) try: - async for event in self: - if event: - self._dispatch(event) + try: + async for event in self: + if event: + self._dispatch(event) + except BaseException as error: + if not connected_once: + LOGGER.error( + "Listen task ended before its first connect: %r", error + ) + raise + if not connected_once: + LOGGER.error("Listen task ended before its first connect") finally: + remove_connection_listener() self._close_response() if self._listen_task is current_task: self._listen_task = None diff --git a/tests/test_listen_early_exit.py b/tests/test_listen_early_exit.py new file mode 100644 index 0000000..f097f50 --- /dev/null +++ b/tests/test_listen_early_exit.py @@ -0,0 +1,178 @@ +"""Regression tests for listen() dying before its first successful connect. + +A listen task that fails before ever reaching a connect must not vanish +quietly: it is the failure mode behind the incident that motivated this - +a listen task that never logged a single connect attempt for 2.5+ days, +discoverable only by an absence in server-side logs. Any death before the +first connect must say so loudly, at ERROR, in the caller's own logs, and +must not swallow the underlying exception. +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from typing import Any + +import aiohttp + +from teslemetry_stream.exception import TeslemetryStreamAuthenticationError +from teslemetry_stream.stream import LOGGER, TeslemetryStream + +REQUEST_INFO = aiohttp.RequestInfo( + url="https://fake.teslemetry.com/sse", + method="GET", + headers={}, # type: ignore[arg-type] + real_url="https://fake.teslemetry.com/sse", # type: ignore[arg-type] +) + + +class BlockingSession: + """A session whose GET never returns until the caller is cancelled - + the shape of a connect() that is stuck (DNS, TCP handshake, proxy).""" + + def __init__(self) -> None: + self.calls = 0 + + async def get(self, url: str, **kwargs: Any) -> Any: + self.calls += 1 + await asyncio.Future() # blocks until this awaiter is cancelled + + +class FakeSession: + """A session whose single `get()` raises the given exception, mirroring + a 401/403 raising straight out of the connect GET.""" + + def __init__(self, exc: BaseException) -> None: + self.calls = 0 + self._exc = exc + + async def get(self, url: str, **kwargs: Any) -> Any: + self.calls += 1 + raise self._exc + + +def make_stream(session: Any) -> TeslemetryStream: + return TeslemetryStream( + session=session, + access_token="test-token", + server="api.teslemetry.com", + manual=True, + ) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def drain_cancelled(task: asyncio.Task[Any]) -> None: + with contextlib.suppress(asyncio.CancelledError): + await task + + +class CaptureLogs: + """Minimal stand-in for pytest's caplog fixture, scoped to LOGGER.""" + + def __enter__(self) -> list[logging.LogRecord]: + self._records: list[logging.LogRecord] = [] + self._handler = _ListHandler(self._records) + self._prev_level = LOGGER.level + LOGGER.addHandler(self._handler) + LOGGER.setLevel(logging.DEBUG) + return self._records + + def __exit__(self, *exc_info: Any) -> None: + LOGGER.removeHandler(self._handler) + LOGGER.setLevel(self._prev_level) + + +class _ListHandler(logging.Handler): + def __init__(self, records: list[logging.LogRecord]) -> None: + super().__init__() + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + self._records.append(record) + + +def early_exit_errors(records: list[logging.LogRecord]) -> list[logging.LogRecord]: + return [ + r + for r in records + if r.levelno == logging.ERROR and "before its first connect" in r.getMessage() + ] + + +async def test_cancelled_before_first_connect_logs_error(results: list[bool]) -> None: + """A stuck connect that is cancelled before it ever succeeds must log + ERROR and re-raise the CancelledError - never vanish silently.""" + session = BlockingSession() + stream = make_stream(session) + + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + + task.cancel() + await drain_cancelled(task) + + results.append( + check( + "cancellation before first connect logs at ERROR", + len(early_exit_errors(records)) == 1, + f"got {early_exit_errors(records)}", + ) + ) + results.append(check("the task ended up cancelled, not swallowed", task.cancelled())) + + +async def test_auth_failure_before_first_connect_logs_error(results: list[bool]) -> None: + """A 401 on the very first connect attempt both surfaces its usual + TeslemetryStreamAuthenticationError and, since it happened before any + connect ever succeeded, is loudly flagged as an early exit.""" + error = aiohttp.ClientResponseError( + request_info=REQUEST_INFO, history=(), status=401, message="Unauthorized" + ) + session = FakeSession(error) + stream = make_stream(session) + + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + + raised: BaseException | None = None + try: + await task + except TeslemetryStreamAuthenticationError as exc: + raised = exc + + results.append( + check( + "the auth error is still re-raised, not swallowed", + isinstance(raised, TeslemetryStreamAuthenticationError), + f"got {raised!r}", + ) + ) + results.append( + check( + "the failure before first connect logs at ERROR", + len(early_exit_errors(records)) == 1, + f"got {early_exit_errors(records)}", + ) + ) + + +async def main() -> None: + results: list[bool] = [] + await test_cancelled_before_first_connect_logs_error(results) + await test_auth_failure_before_first_connect_logs_error(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()) diff --git a/tests/test_silence_watchdog.py b/tests/test_silence_watchdog.py new file mode 100644 index 0000000..9d546d8 --- /dev/null +++ b/tests/test_silence_watchdog.py @@ -0,0 +1,251 @@ +"""Regression tests for the silence watchdog in `__anext__`: a connection +that stops delivering any bytes - not even an SSE keepalive comment - for +longer than the silence window is closed and reconnected by the existing +loop, instead of hanging forever like the customer incident that motivated +this (a Home Assistant instance whose stream connection went dead for 2.5+ +days with no reconnect attempt). +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from typing import Any, Callable + +import teslemetry_stream.stream as stream_module +from teslemetry_stream.stream import LOGGER, TeslemetryStream + + +class QueueContent: + """Async-iterable response body fed lines on the test's own timeline.""" + + def __init__(self) -> None: + self._queue: asyncio.Queue[bytes] = asyncio.Queue() + + def __aiter__(self) -> QueueContent: + return self + + async def __anext__(self) -> bytes: + return await self._queue.get() + + def push(self, line: bytes) -> None: + self._queue.put_nowait(line) + + +class FakeResponse: + def __init__(self) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + self.content = QueueContent() + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeSession: + def __init__(self) -> None: + self.calls = 0 + self.responses: list[FakeResponse] = [] + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls += 1 + response = FakeResponse() + self.responses.append(response) + return response + + +def make_stream(session: FakeSession) -> TeslemetryStream: + return TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + manual=True, + ) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def drain_cancelled(task: asyncio.Task[Any]) -> None: + with contextlib.suppress(asyncio.CancelledError): + await task + + +class CaptureLogs: + """Minimal stand-in for pytest's caplog fixture, scoped to LOGGER.""" + + def __enter__(self) -> list[logging.LogRecord]: + self._records: list[logging.LogRecord] = [] + self._handler = _ListHandler(self._records) + self._prev_level = LOGGER.level + LOGGER.addHandler(self._handler) + LOGGER.setLevel(logging.DEBUG) + return self._records + + def __exit__(self, *exc_info: Any) -> None: + LOGGER.removeHandler(self._handler) + LOGGER.setLevel(self._prev_level) + + +class _ListHandler(logging.Handler): + def __init__(self, records: list[logging.LogRecord]) -> None: + super().__init__() + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + self._records.append(record) + + +def with_silence_window(window: float) -> Callable[[], None]: + """Patch the module-level silence window; returns a restore function.""" + original = stream_module._SILENCE_WINDOW + stream_module._SILENCE_WINDOW = window + + def restore() -> None: + stream_module._SILENCE_WINDOW = original + + return restore + + +async def test_silence_past_window_triggers_reconnect(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session) + restore = with_silence_window(0.1) + + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append( + check("initial connect happened", session.calls == 1, f"got {session.calls}") + ) + + await asyncio.sleep(0.15) # past the (patched) silence window, before a second one + + results.append( + check( + "a reconnect attempt was made after silence", + session.calls == 2, + f"got {session.calls}", + ) + ) + results.append(check("the silent response was closed", session.responses[0].closed)) + + warning_records = [ + r for r in records if r.levelno == logging.WARNING and "No data received" in r.getMessage() + ] + results.append( + check( + "the first silence-triggered reconnect logs at WARNING", + len(warning_records) == 1, + f"got {warning_records}", + ) + ) + + restore() + stream.close() + await drain_cancelled(task) + + +async def test_repeated_silence_logs_debug_after_first_warning(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session) + restore = with_silence_window(0.1) + + with CaptureLogs() as records: + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + + # Two consecutive silence windows with no successful read in between. + await asyncio.sleep(0.15) + await asyncio.sleep(0.1) + + results.append( + check( + "a second reconnect attempt was made", + session.calls == 3, + f"got {session.calls}", + ) + ) + + warning_records = [ + r for r in records if r.levelno == logging.WARNING and "No data received" in r.getMessage() + ] + debug_records = [ + r for r in records if r.levelno == logging.DEBUG and "No data received" in r.getMessage() + ] + results.append( + check( + "only the first silence-triggered reconnect logs at WARNING", + len(warning_records) == 1, + f"got {warning_records}", + ) + ) + results.append( + check( + "the repeat silence-triggered reconnect logs at DEBUG", + len(debug_records) == 1, + f"got {debug_records}", + ) + ) + + restore() + stream.close() + await drain_cancelled(task) + + +async def test_keepalive_within_window_does_not_reconnect(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session) + restore = with_silence_window(0.2) + + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append( + check("initial connect happened", session.calls == 1, f"got {session.calls}") + ) + + # A keepalive comment line, well inside the window, should reset it. + await asyncio.sleep(0.1) + session.responses[0].content.push(b": keepalive\n") + await asyncio.sleep(0) + + # Elapsed since connect (0.1 + 0.1 = 0.2) would exceed the window if it + # hadn't been reset by the keepalive above. + await asyncio.sleep(0.1) + + results.append( + check( + "no reconnect happens while keepalives arrive inside the window", + session.calls == 1, + f"got {session.calls}", + ) + ) + results.append( + check("the connection stays open", not session.responses[0].closed) + ) + + restore() + stream.close() + await drain_cancelled(task) + + +async def main() -> None: + results: list[bool] = [] + await test_silence_past_window_triggers_reconnect(results) + await test_repeated_silence_logs_debug_after_first_warning(results) + await test_keepalive_within_window_does_not_reconnect(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()) From 2620120b2526f2c84530511f43dd3ba559f5a30b Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 15 Sep 2026 08:42:43 +1000 Subject: [PATCH 2/4] fix: give version-sensitive stream tests enough ticks for wait_for's overhead Wrapping each SSE line read in asyncio.wait_for (the silence watchdog) adds a couple of extra event-loop hops before a completed read is observable on Python <3.12, where wait_for still allocates a bridging waiter future instead of the 3.12+ fast path. Several existing tests advanced the loop by a fixed, version-tuned number of sleep(0) ticks that no longer covered this; give them enough margin to pass on every supported Python version. No production behavior changes - the clean-end path still logs INFO and reconnects immediately. --- tests/test_clean_stream_end.py | 7 +++++-- tests/test_stream_lifecycle.py | 21 +++++++++++++++++---- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/tests/test_clean_stream_end.py b/tests/test_clean_stream_end.py index 010d6f8..96a5929 100644 --- a/tests/test_clean_stream_end.py +++ b/tests/test_clean_stream_end.py @@ -102,8 +102,11 @@ async def test_clean_end_logs_info_and_reconnects(results: list[bool]) -> None: ) session.responses[0].content.end() - await asyncio.sleep(0) - await asyncio.sleep(0) + # asyncio.wait_for's pre-3.12 implementation adds a couple of extra + # scheduling hops before a completed read is observable - enough + # ticks to cover every supported Python version. + for _ in range(6): + await asyncio.sleep(0) results.append( check( diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 03c2fa5..7e76b10 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -255,7 +255,11 @@ async def test_close_prevents_reconnect_after_backoff(results: list[bool]) -> No results.append(check("initial connect happened", session.calls == 1)) session.responses[0].content.fail(aiohttp.ClientError("boom")) - await asyncio.sleep(0) # let __anext__ observe the error and enter the backoff sleep + # asyncio.wait_for's pre-3.12 implementation adds a couple of extra + # scheduling hops before a completed read is observable - a handful of + # sleep(0) ticks covers every supported Python version. + for _ in range(6): + await asyncio.sleep(0) results.append( check("the failed response was closed before backoff", session.responses[0].closed) @@ -360,7 +364,10 @@ def mutate_during_dispatch(event: dict[str, Any]) -> None: stream.async_add_listener(mutate_during_dispatch, {"vin": None}) - for _ in range(5): + # asyncio.wait_for's pre-3.12 implementation adds a couple of extra + # scheduling hops before a completed read is observable - enough ticks + # to cover every supported Python version. + for _ in range(12): await asyncio.sleep(0) results.append( @@ -416,7 +423,10 @@ def public_mutator(event: dict[str, Any]) -> None: stream.async_add_listener(public_mutator) vehicle = TeslemetryStreamVehicle(stream, VIN) - for _ in range(5): + # asyncio.wait_for's pre-3.12 implementation adds a couple of extra + # scheduling hops before a completed read is observable - enough ticks + # to cover every supported Python version. + for _ in range(12): await asyncio.sleep(0) results.append( @@ -464,7 +474,10 @@ def generic_listener(event: dict[str, Any]) -> None: stream.async_add_listener(generic_listener, {"vin": None}) - for _ in range(5): + # asyncio.wait_for's pre-3.12 implementation adds a couple of extra + # scheduling hops before a completed read is observable - enough ticks + # to cover every supported Python version. + for _ in range(12): await asyncio.sleep(0) results.append(check("the listener discovered the new vehicle", len(discovered) == 1)) From e42f892d4c0a71aff4781472adf79cdf7c03c93a Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 15 Sep 2026 09:17:12 +1000 Subject: [PATCH 3/4] fix: simplify to a single WARNING on early listen() exit, drop silence watchdog sock_read=30 on the connect() request already bounds a connection that stops producing bytes (data or SSE keepalive) and routes into the existing ClientError handler, which already logs and reconnects - the added wait_for-based watchdog was redundant complexity for no new behavior. Restore sock_read=30 and drop the wait_for wrapper and streak logging. Keep only the early-exit change: listen() logs one WARNING line if it returns or raises before its first successful connect, and still re-raises rather than swallowing the failure. --- AGENTS.md | 5 +- teslemetry_stream/stream.py | 44 +----- tests/test_clean_stream_end.py | 7 +- tests/test_listen_early_exit.py | 10 +- tests/test_silence_watchdog.py | 251 -------------------------------- tests/test_stream_lifecycle.py | 21 +-- 6 files changed, 18 insertions(+), 320 deletions(-) delete mode 100644 tests/test_silence_watchdog.py diff --git a/AGENTS.md b/AGENTS.md index c14c94e..bd29a13 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,8 +24,8 @@ Project-intrinsic knowledge that should travel with the code: build, test, relea - Internal reconnect paths (EOF, `ClientError`, unexpected exceptions) call `_close_response()`, **never** `close()` - `close()` additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally so cancellation still releases the connection. - Dispatch iterates a **sorted snapshot** of `_listeners.values()` with internal listeners first, never the live dict: a callback adding a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration`, and a public callback must not mutate the event before an internal bookkeeping listener has cached from it. `_update_connection_listeners()` has the same mutation hazard and iterates a plain `list(...)` snapshot (no ordering requirement - connection listeners share no mutable event). - `__anext__` treats `ClientResponseError` with status 401/403 as terminal: sets `active = False` and raises `TeslemetryStreamAuthenticationError`. Every other `aiohttp.ClientError` keeps backoff-and-reconnect. -- `__anext__` reads each SSE line under `asyncio.wait_for(..., timeout=_SILENCE_WINDOW)` - this is the one mechanism bounding a connection gone quiet (no data, no keepalive) without erroring; `connect()`'s `sock_read` is deliberately unbounded so the two don't race. A timeout closes the response and falls through to the outer loop's immediate reconnect, same as a clean server-side end. `_silence_streak` gates logging: the first silence-triggered reconnect after a real read logs WARNING, further ones with still no read in between log DEBUG - it resets only on a successfully read line (data or keepalive), not on a bare reconnect. -- `listen()` tracks whether it has ever seen a successful connect (via a scoped internal connection listener). If it returns or raises before that - including cancellation - it logs ERROR before propagating; nothing before a first connect is swallowed. +- `connect()`'s `sock_read=30` bounds a connection that stops producing bytes (no data, no SSE keepalive) - a read timeout raises `aiohttp.ClientError`, which the existing client-error handler in `__anext__` already logs and reconnects. There is no separate silence mechanism. +- `listen()` tracks whether it has ever seen a successful connect (via a scoped internal connection listener). If it returns or raises before that - including cancellation - it logs one WARNING line before propagating; nothing before a first connect is swallowed. - `async_add_listener(..., internal=True)` marks a bookkeeping-only listener (the vehicle config-sync listener is the only user). Its `schedule_refresh` gate is unconditionally false, which is what makes construction-time registration safe outside a running event loop. Both the "first listener starts the task" and "last listener removed auto-closes" checks count only public listeners, so an internal listener can neither pin the connection open nor block a later public listener's zero-to-one restart. **Any stream stand-in (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`.** - `topics=` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param. `SseTopic` is the closed server-recognized set; `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets, flat per-product-kind lists deliberately not split by whether a topic has a connect-time snapshot (server behavior, not something this library encodes). Omitting it (`None`) is legacy-all forever. An explicitly empty iterable raises `ValueError` at construction - "no topics" must not mean "all topics", mirroring the server's 400. A bare `str`/`SseTopic` is accepted as a single topic, not iterated character-by-character; `topics: str | Iterable[str] | None` exists precisely because a lone string also satisfies `Iterable[str]`. - `ingest()` (and `TeslemetryStreamVehicle.ingest()`, same call with the VIN filled in) is the entry point for an observation not read off the SSE connection - a Bluetooth broadcast, today. It builds the native wire event (`vin`/`data`/`createdAt`) plus an open-ended `metadata` dict (`Metadata.SOURCE`/`Metadata.RAW`) and hands it to the same `_dispatch` `listen()` uses, so existing `listen_*` callbacks receive both sources with no translation and no second subscription. Dispatch is arrival-ordered and the stream holds no per-field value: **nothing is deduplicated, reordered, or dropped, and there is deliberately no source ranking or precedence** - which report to believe is the consumer's decision, made on `metadata`. Ingesting neither requires nor opens a connection. Neither library depends on the other: the BLE-side shim shaping a broadcast into this format lives in `tesla-fleet-api` and the consumer wires the two. @@ -58,7 +58,6 @@ Project-intrinsic knowledge that should travel with the code: build, test, relea | Listen task ownership, dispatch snapshots/order, close races | `test_stream_lifecycle.py` | | 401/403 terminal vs transient retry | `test_auth_failure.py` | | Clean vs abrupt stream end logging | `test_clean_stream_end.py` | -| Silence watchdog: timeout reconnect, keepalive reset, WARNING-once/DEBUG-repeat | `test_silence_watchdog.py` | | `listen()` early exit before first connect logs ERROR and re-raises | `test_listen_early_exit.py` | | `str` payload coercion against real telemetry | `test_field_type_coercion.py` | | `topics` URL construction, empty/bare-string handling, tariff listener | `test_sse_topics.py` | diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index e7835f9..4046a7c 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -16,11 +16,6 @@ LOGGER = logging.getLogger(__package__) -# The server sends an SSE keepalive comment every 10s. Three missed -# keepalives (30s of total silence) bounds reconnecting a connection that -# has gone quiet without erroring. -_SILENCE_WINDOW = 30.0 - class TeslemetryStream: """Teslemetry Stream Client""" @@ -77,9 +72,6 @@ def __init__( ] = {} self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {} self._listen_task: asyncio.Task[None] | None = None - # Consecutive silence-triggered reconnects with no successful read in - # between - gates the WARNING-once/DEBUG-after-that logging split. - self._silence_streak: int = 0 # Created lazily in connect() - asyncio.Lock() requires a running # loop on Python 3.9, and streams are commonly built before one. self._connect_lock: asyncio.Lock | None = None @@ -268,12 +260,7 @@ async def connect(self) -> None: headers=headers, params=params, raise_for_status=True, - # sock_read is intentionally unbounded: the silence watchdog - # in __anext__ is the one mechanism that bounds dead air on - # an open connection, with logging this bare timeout can't do. - timeout=aiohttp.ClientTimeout( - connect=5, sock_connect=5, sock_read=None, total=None - ), + timeout=aiohttp.ClientTimeout(connect=5, sock_connect=5, sock_read=30, total=None), chunked=True, ) if not self.active: @@ -345,33 +332,14 @@ async def __anext__(self) -> dict[str, Any]: # Connect to the stream await self.connect() assert self._response - content_iter = self._response.content.__aiter__() - while True: - try: - line_in_bytes = await asyncio.wait_for( - content_iter.__anext__(), timeout=_SILENCE_WINDOW - ) - except StopAsyncIteration: - raise TeslemetryStreamEnded() from None - except asyncio.TimeoutError: - if self._silence_streak == 0: - LOGGER.warning( - "No data received for %ss, reconnecting", _SILENCE_WINDOW - ) - else: - LOGGER.debug( - "No data received for %ss, reconnecting", _SILENCE_WINDOW - ) - self._silence_streak += 1 - self._close_response() - break - self._silence_streak = 0 + async for line_in_bytes in self._response.content: field, _, value = line_in_bytes.decode("utf8").partition(": ") if field == "data": data = json.loads(value) if self.parse_timestamp: data["timestamp"] = _parse_created_at(data["createdAt"]) return cast(dict[str, Any], data) + raise TeslemetryStreamEnded() except StopAsyncIteration as e: # Re-raise explicitly so it isn't caught by the generic Exception handler below self.disconnect() @@ -490,12 +458,10 @@ def _mark_connected(value: bool) -> None: self._dispatch(event) except BaseException as error: if not connected_once: - LOGGER.error( - "Listen task ended before its first connect: %r", error - ) + LOGGER.warning("Listen task ended before its first connect: %r", error) raise if not connected_once: - LOGGER.error("Listen task ended before its first connect") + LOGGER.warning("Listen task ended before its first connect") finally: remove_connection_listener() self._close_response() diff --git a/tests/test_clean_stream_end.py b/tests/test_clean_stream_end.py index 96a5929..010d6f8 100644 --- a/tests/test_clean_stream_end.py +++ b/tests/test_clean_stream_end.py @@ -102,11 +102,8 @@ async def test_clean_end_logs_info_and_reconnects(results: list[bool]) -> None: ) session.responses[0].content.end() - # asyncio.wait_for's pre-3.12 implementation adds a couple of extra - # scheduling hops before a completed read is observable - enough - # ticks to cover every supported Python version. - for _ in range(6): - await asyncio.sleep(0) + await asyncio.sleep(0) + await asyncio.sleep(0) results.append( check( diff --git a/tests/test_listen_early_exit.py b/tests/test_listen_early_exit.py index f097f50..8f2b145 100644 --- a/tests/test_listen_early_exit.py +++ b/tests/test_listen_early_exit.py @@ -4,7 +4,7 @@ quietly: it is the failure mode behind the incident that motivated this - a listen task that never logged a single connect attempt for 2.5+ days, discoverable only by an absence in server-side logs. Any death before the -first connect must say so loudly, at ERROR, in the caller's own logs, and +first connect must say so loudly, at WARNING, in the caller's own logs, and must not swallow the underlying exception. """ from __future__ import annotations @@ -100,13 +100,13 @@ def early_exit_errors(records: list[logging.LogRecord]) -> list[logging.LogRecor return [ r for r in records - if r.levelno == logging.ERROR and "before its first connect" in r.getMessage() + if r.levelno == logging.WARNING and "before its first connect" in r.getMessage() ] async def test_cancelled_before_first_connect_logs_error(results: list[bool]) -> None: """A stuck connect that is cancelled before it ever succeeds must log - ERROR and re-raise the CancelledError - never vanish silently.""" + WARNING and re-raise the CancelledError - never vanish silently.""" session = BlockingSession() stream = make_stream(session) @@ -120,7 +120,7 @@ async def test_cancelled_before_first_connect_logs_error(results: list[bool]) -> results.append( check( - "cancellation before first connect logs at ERROR", + "cancellation before first connect logs at WARNING", len(early_exit_errors(records)) == 1, f"got {early_exit_errors(records)}", ) @@ -156,7 +156,7 @@ async def test_auth_failure_before_first_connect_logs_error(results: list[bool]) ) results.append( check( - "the failure before first connect logs at ERROR", + "the failure before first connect logs at WARNING", len(early_exit_errors(records)) == 1, f"got {early_exit_errors(records)}", ) diff --git a/tests/test_silence_watchdog.py b/tests/test_silence_watchdog.py deleted file mode 100644 index 9d546d8..0000000 --- a/tests/test_silence_watchdog.py +++ /dev/null @@ -1,251 +0,0 @@ -"""Regression tests for the silence watchdog in `__anext__`: a connection -that stops delivering any bytes - not even an SSE keepalive comment - for -longer than the silence window is closed and reconnected by the existing -loop, instead of hanging forever like the customer incident that motivated -this (a Home Assistant instance whose stream connection went dead for 2.5+ -days with no reconnect attempt). -""" -from __future__ import annotations - -import asyncio -import contextlib -import logging -from typing import Any, Callable - -import teslemetry_stream.stream as stream_module -from teslemetry_stream.stream import LOGGER, TeslemetryStream - - -class QueueContent: - """Async-iterable response body fed lines on the test's own timeline.""" - - def __init__(self) -> None: - self._queue: asyncio.Queue[bytes] = asyncio.Queue() - - def __aiter__(self) -> QueueContent: - return self - - async def __anext__(self) -> bytes: - return await self._queue.get() - - def push(self, line: bytes) -> None: - self._queue.put_nowait(line) - - -class FakeResponse: - def __init__(self) -> None: - self.url = "https://fake.teslemetry.com/sse" - self.status = 200 - self.content = QueueContent() - self.closed = False - - def close(self) -> None: - self.closed = True - - -class FakeSession: - def __init__(self) -> None: - self.calls = 0 - self.responses: list[FakeResponse] = [] - - async def get(self, url: str, **kwargs: Any) -> FakeResponse: - self.calls += 1 - response = FakeResponse() - self.responses.append(response) - return response - - -def make_stream(session: FakeSession) -> TeslemetryStream: - return TeslemetryStream( - session=session, # type: ignore[arg-type] - access_token="test-token", - server="api.teslemetry.com", - manual=True, - ) - - -def check(label: str, ok: bool, detail: str = "") -> bool: - print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") - return ok - - -async def drain_cancelled(task: asyncio.Task[Any]) -> None: - with contextlib.suppress(asyncio.CancelledError): - await task - - -class CaptureLogs: - """Minimal stand-in for pytest's caplog fixture, scoped to LOGGER.""" - - def __enter__(self) -> list[logging.LogRecord]: - self._records: list[logging.LogRecord] = [] - self._handler = _ListHandler(self._records) - self._prev_level = LOGGER.level - LOGGER.addHandler(self._handler) - LOGGER.setLevel(logging.DEBUG) - return self._records - - def __exit__(self, *exc_info: Any) -> None: - LOGGER.removeHandler(self._handler) - LOGGER.setLevel(self._prev_level) - - -class _ListHandler(logging.Handler): - def __init__(self, records: list[logging.LogRecord]) -> None: - super().__init__() - self._records = records - - def emit(self, record: logging.LogRecord) -> None: - self._records.append(record) - - -def with_silence_window(window: float) -> Callable[[], None]: - """Patch the module-level silence window; returns a restore function.""" - original = stream_module._SILENCE_WINDOW - stream_module._SILENCE_WINDOW = window - - def restore() -> None: - stream_module._SILENCE_WINDOW = original - - return restore - - -async def test_silence_past_window_triggers_reconnect(results: list[bool]) -> None: - session = FakeSession() - stream = make_stream(session) - restore = with_silence_window(0.1) - - with CaptureLogs() as records: - task = asyncio.create_task(stream.listen()) - await asyncio.sleep(0) - await asyncio.sleep(0) - results.append( - check("initial connect happened", session.calls == 1, f"got {session.calls}") - ) - - await asyncio.sleep(0.15) # past the (patched) silence window, before a second one - - results.append( - check( - "a reconnect attempt was made after silence", - session.calls == 2, - f"got {session.calls}", - ) - ) - results.append(check("the silent response was closed", session.responses[0].closed)) - - warning_records = [ - r for r in records if r.levelno == logging.WARNING and "No data received" in r.getMessage() - ] - results.append( - check( - "the first silence-triggered reconnect logs at WARNING", - len(warning_records) == 1, - f"got {warning_records}", - ) - ) - - restore() - stream.close() - await drain_cancelled(task) - - -async def test_repeated_silence_logs_debug_after_first_warning(results: list[bool]) -> None: - session = FakeSession() - stream = make_stream(session) - restore = with_silence_window(0.1) - - with CaptureLogs() as records: - task = asyncio.create_task(stream.listen()) - await asyncio.sleep(0) - await asyncio.sleep(0) - - # Two consecutive silence windows with no successful read in between. - await asyncio.sleep(0.15) - await asyncio.sleep(0.1) - - results.append( - check( - "a second reconnect attempt was made", - session.calls == 3, - f"got {session.calls}", - ) - ) - - warning_records = [ - r for r in records if r.levelno == logging.WARNING and "No data received" in r.getMessage() - ] - debug_records = [ - r for r in records if r.levelno == logging.DEBUG and "No data received" in r.getMessage() - ] - results.append( - check( - "only the first silence-triggered reconnect logs at WARNING", - len(warning_records) == 1, - f"got {warning_records}", - ) - ) - results.append( - check( - "the repeat silence-triggered reconnect logs at DEBUG", - len(debug_records) == 1, - f"got {debug_records}", - ) - ) - - restore() - stream.close() - await drain_cancelled(task) - - -async def test_keepalive_within_window_does_not_reconnect(results: list[bool]) -> None: - session = FakeSession() - stream = make_stream(session) - restore = with_silence_window(0.2) - - task = asyncio.create_task(stream.listen()) - await asyncio.sleep(0) - await asyncio.sleep(0) - results.append( - check("initial connect happened", session.calls == 1, f"got {session.calls}") - ) - - # A keepalive comment line, well inside the window, should reset it. - await asyncio.sleep(0.1) - session.responses[0].content.push(b": keepalive\n") - await asyncio.sleep(0) - - # Elapsed since connect (0.1 + 0.1 = 0.2) would exceed the window if it - # hadn't been reset by the keepalive above. - await asyncio.sleep(0.1) - - results.append( - check( - "no reconnect happens while keepalives arrive inside the window", - session.calls == 1, - f"got {session.calls}", - ) - ) - results.append( - check("the connection stays open", not session.responses[0].closed) - ) - - restore() - stream.close() - await drain_cancelled(task) - - -async def main() -> None: - results: list[bool] = [] - await test_silence_past_window_triggers_reconnect(results) - await test_repeated_silence_logs_debug_after_first_warning(results) - await test_keepalive_within_window_does_not_reconnect(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()) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 7e76b10..03c2fa5 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -255,11 +255,7 @@ async def test_close_prevents_reconnect_after_backoff(results: list[bool]) -> No results.append(check("initial connect happened", session.calls == 1)) session.responses[0].content.fail(aiohttp.ClientError("boom")) - # asyncio.wait_for's pre-3.12 implementation adds a couple of extra - # scheduling hops before a completed read is observable - a handful of - # sleep(0) ticks covers every supported Python version. - for _ in range(6): - await asyncio.sleep(0) + await asyncio.sleep(0) # let __anext__ observe the error and enter the backoff sleep results.append( check("the failed response was closed before backoff", session.responses[0].closed) @@ -364,10 +360,7 @@ def mutate_during_dispatch(event: dict[str, Any]) -> None: stream.async_add_listener(mutate_during_dispatch, {"vin": None}) - # asyncio.wait_for's pre-3.12 implementation adds a couple of extra - # scheduling hops before a completed read is observable - enough ticks - # to cover every supported Python version. - for _ in range(12): + for _ in range(5): await asyncio.sleep(0) results.append( @@ -423,10 +416,7 @@ def public_mutator(event: dict[str, Any]) -> None: stream.async_add_listener(public_mutator) vehicle = TeslemetryStreamVehicle(stream, VIN) - # asyncio.wait_for's pre-3.12 implementation adds a couple of extra - # scheduling hops before a completed read is observable - enough ticks - # to cover every supported Python version. - for _ in range(12): + for _ in range(5): await asyncio.sleep(0) results.append( @@ -474,10 +464,7 @@ def generic_listener(event: dict[str, Any]) -> None: stream.async_add_listener(generic_listener, {"vin": None}) - # asyncio.wait_for's pre-3.12 implementation adds a couple of extra - # scheduling hops before a completed read is observable - enough ticks - # to cover every supported Python version. - for _ in range(12): + for _ in range(5): await asyncio.sleep(0) results.append(check("the listener discovered the new vehicle", len(discovered) == 1)) From a582535b71fc67addc1c8030ac5d6ea11d4fa588 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 15 Sep 2026 09:22:38 +1000 Subject: [PATCH 4/4] fix: restore byte-identical ClientTimeout formatting Keep the PR diff scoped to the listen() early-exit change only. --- teslemetry_stream/stream.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 4046a7c..35c1305 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -260,7 +260,9 @@ async def connect(self) -> None: headers=headers, params=params, raise_for_status=True, - timeout=aiohttp.ClientTimeout(connect=5, sock_connect=5, sock_read=30, total=None), + timeout=aiohttp.ClientTimeout( + connect=5, sock_connect=5, sock_read=30, total=None + ), chunked=True, ) if not self.active: