diff --git a/AGENTS.md b/AGENTS.md index 9472246..5619d3a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design), so it can't actually promise the whole REST-shaped document; a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. - `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) 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 (must stay in sync with the api's `SSE_TOPICS`), 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 - and existing callers that never pass it are unaffected. 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. +- `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 3ba6988..c96fbc6 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -69,6 +69,10 @@ def __init__( Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None] ] = {} self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {} + self._listen_task: asyncio.Task[None] | None = None + # 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 self._session = session self.access_token = access_token self.parse_timestamp = parse_timestamp @@ -230,44 +234,72 @@ async def connect(self) -> None: if not self.server: await self.get_config() - LOGGER.debug("Connecting to %s", self.server) - url = f"https://{self.server}/sse" - if self.vin: - url += f"/{self.vin}" - headers = await self.headers() - params = {"topics": ",".join(self.topics)} if self.topics else None - self._response = await self._session.get( - url, - headers=headers, - params=params, - raise_for_status=True, - timeout=aiohttp.ClientTimeout( - connect=5, sock_connect=5, sock_read=30, total=None - ), - chunked=True, - ) - LOGGER.debug( - "Connected to %s with status %s", self._response.url, self._response.status - ) - self.retries = 0 - self._update_connection_listeners(True) + if self._connect_lock is None: + self._connect_lock = asyncio.Lock() + async with self._connect_lock: + if not self.active: + # Stopped while waiting for the lock; a concurrent caller may + # already be connected, or a stop was requested outright. + return + + LOGGER.debug("Connecting to %s", self.server) + url = f"https://{self.server}/sse" + if self.vin: + url += f"/{self.vin}" + headers = await self.headers() + params = {"topics": ",".join(self.topics)} if self.topics else None + response = await self._session.get( + url, + headers=headers, + params=params, + raise_for_status=True, + timeout=aiohttp.ClientTimeout( + connect=5, sock_connect=5, sock_read=30, total=None + ), + chunked=True, + ) + if not self.active: + # Stopped while the request was in flight - discard it rather + # than publish a response nobody will ever close. + response.close() + return + if self._response is not None: + self._response.close() + self._response = response + LOGGER.debug( + "Connected to %s with status %s", self._response.url, self._response.status + ) + self.retries = 0 + self._update_connection_listeners(True) def disconnect(self) -> None: """ Disconnect from the telemetry stream. """ - self.active = False self.close() - def close(self) -> None: + def _close_response(self) -> None: """ - Close connection. + Close the current response, if any, without changing whether the + stream is meant to keep running - used by reconnect paths and by + listen()'s cleanup, as opposed to close()'s full stop. """ if self._response is not None: LOGGER.debug("Disconnecting from %s", self.server) self._response.close() self._response = None - self._update_connection_listeners(False) + self._update_connection_listeners(False) + + def close(self) -> None: + """ + Stop the stream: closes the response and cancels the owned listen + task so a running listener does not immediately reconnect. + """ + self.active = False + task, self._listen_task = self._listen_task, None + if task is not None and not task.done(): + task.cancel() + self._close_response() def __aiter__(self) -> TeslemetryStream: """ @@ -312,17 +344,17 @@ async def __anext__(self) -> dict[str, Any]: raise e except TeslemetryStreamEnded: LOGGER.warning("Stream ended by server") - self.close() + self._close_response() except aiohttp.ClientError as error: LOGGER.warning("Client error: %s", repr(error)) - self.close() + self._close_response() delay = min(2**self.retries, 600) LOGGER.debug("Reconnecting in %s seconds", delay) await asyncio.sleep(delay) self.retries += 1 except Exception as error: LOGGER.error("Unexpected error: %s", repr(error)) - self.close() + self._close_response() LOGGER.debug("Reconnecting in %s seconds", 1) await asyncio.sleep(1) @@ -347,28 +379,52 @@ def remove_listener() -> None: self._listeners.pop(remove_listener) if not self._listeners: LOGGER.info("Shutting down stream as there are no more listeners") - self.active = False + self.close() self._listeners[remove_listener] = (callback, filters) - # This is the first listener, set up task. - if schedule_refresh and not self.manual: - asyncio.create_task(self.listen()) + # This is the first listener - start the owned listen task, unless + # one is already running or manual mode delegates that to the caller. + if ( + schedule_refresh + and not self.manual + and (self._listen_task is None or self._listen_task.done()) + ): + self._listen_task = asyncio.create_task(self.listen()) return remove_listener async def listen(self) -> None: """ Listen to the telemetry stream. - """ - async for event in self: - if event: - for listener, filters in self._listeners.values(): - if recursive_match(filters, event): - try: - listener(event) - except Exception as error: - LOGGER.error("Uncaught error in listener: %s", error) + + A second concurrent call joins the already-running owned task + instead of starting a competing reader on the same connection. + """ + current_task = asyncio.current_task() + existing_task = self._listen_task + if ( + existing_task is not None + and not existing_task.done() + and existing_task is not current_task + ): + await existing_task + return + + self._listen_task = current_task + try: + async for event in self: + if event: + for listener, filters in self._listeners.values(): + if 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: + self._listen_task = None LOGGER.debug("Listen has finished") def listen_Credits( diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py new file mode 100644 index 0000000..2b0cf47 --- /dev/null +++ b/tests/test_stream_lifecycle.py @@ -0,0 +1,261 @@ +"""Regression tests for the SSE stream lifecycle hardening: an owned listen +task, serialized connect(), and close() as a real stop rather than a +response-close-only operation. + +Scenarios mirror the leak audit's proven defects: +- no listener/connect single-flight (concurrent connects overwrite the one + `_response` slot, orphaning the first response); +- last-listener removal only flipped `active`, leaving the response open; +- `close()` didn't stop a running listener or survive task cancellation. +""" +from __future__ import annotations + +import asyncio +import contextlib +from typing import Any + +import aiohttp + +from teslemetry_stream.stream import TeslemetryStream + + +class FakeContent: + """Async-iterable response body that blocks until failed or cancelled.""" + + def __init__(self) -> None: + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeContent: + return self + + async def __anext__(self) -> bytes: + await self._blocker + raise AssertionError("unreachable - blocker only resolves via an exception") + + def fail(self, exc: BaseException) -> None: + if not self._blocker.done(): + self._blocker.set_exception(exc) + + +class FakeResponse: + """Minimal stand-in for the aiohttp response `connect()` awaits.""" + + def __init__(self) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + self.content = FakeContent() + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeSession: + """Captures every `get()` call; optionally gates them on an event.""" + + def __init__(self) -> None: + self.calls = 0 + self.responses: list[FakeResponse] = [] + self.gate: asyncio.Event | None = None + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls += 1 + if self.gate is not None: + await self.gate.wait() + response = FakeResponse() + self.responses.append(response) + return response + + +def make_stream(session: FakeSession, manual: bool = False) -> TeslemetryStream: + return TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + manual=manual, + ) + + +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 + + +async def test_add_remove_readd_before_loop_runs(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session) + + remove1 = stream.async_add_listener(lambda event: None) + task1 = stream._listen_task + remove1() + remove2 = stream.async_add_listener(lambda event: None) + task2 = stream._listen_task + + results.append( + check( + "re-add before the loop runs schedules a distinct owned task", + task2 is not None and task2 is not task1, + ) + ) + + # Let the event loop actually run the (cancelled) first task and the + # (live) second one. + await asyncio.sleep(0) + await asyncio.sleep(0) + + results.append( + check( + "exactly one connect happens despite add/remove/re-add racing the task", + session.calls == 1, + f"got {session.calls}", + ) + ) + results.append(check("the stream ends up connected", stream.connected)) + + stream.close() + await asyncio.sleep(0) + remove2() + + results.append( + check( + "close leaves no open response behind", + bool(session.responses) and all(r.closed for r in session.responses), + ) + ) + + +async def test_two_explicit_listen_calls(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session, manual=True) + + task_a = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append( + check("the first listen() call connects", session.calls == 1, f"got {session.calls}") + ) + + task_b = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + results.append( + check( + "a second concurrent listen() call joins rather than reconnecting", + session.calls == 1, + f"got {session.calls}", + ) + ) + results.append( + check("the owned task stays the first caller's", stream._listen_task is task_a) + ) + + stream.close() + await drain_cancelled(task_a) + await drain_cancelled(task_b) + + results.append( + check("both callers finish once the owner is closed", task_a.done() and task_b.done()) + ) + results.append( + check( + "close leaves no open response behind", + bool(session.responses) and all(r.closed for r in session.responses), + ) + ) + + +async def test_cancel_while_blocked_reading(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session, manual=True) + + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append(check("connected before cancellation", stream.connected and session.calls == 1)) + + # Cancel the task directly - not via stream.close() - to prove listen()'s + # own finally, not an explicit stop call, is what closes the response. + task.cancel() + await drain_cancelled(task) + + results.append( + check( + "cancellation while blocked in content still closes the response", + bool(session.responses) and session.responses[0].closed, + ) + ) + results.append(check("the response reference is cleared", stream._response is None)) + results.append(check("the owned task reference is cleared", stream._listen_task is None)) + + +async def test_close_during_connect(results: list[bool]) -> None: + session = FakeSession() + session.gate = asyncio.Event() + stream = make_stream(session, manual=True) + + task = asyncio.create_task(stream.connect()) + await asyncio.sleep(0) # let connect() reach the gated session.get() + + stream.close() # active=False while the GET is still in flight + session.gate.set() # now let the late response arrive + await task + + results.append( + check("a response arriving after close is not published", stream._response is None) + ) + results.append( + check( + "a response arriving after close is closed, not leaked", + len(session.responses) == 1 and session.responses[0].closed, + ) + ) + + +async def test_close_prevents_reconnect_after_backoff(results: list[bool]) -> None: + session = FakeSession() + stream = make_stream(session, manual=True) + + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + 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 + + results.append( + check("the failed response was closed before backoff", session.responses[0].closed) + ) + + stream.close() # cancel mid-backoff + await drain_cancelled(task) + + results.append( + check( + "no reconnect attempt happens once closed during backoff", + session.calls == 1, + f"got {session.calls}", + ) + ) + + +async def main() -> None: + results: list[bool] = [] + await test_add_remove_readd_before_loop_runs(results) + await test_two_explicit_listen_calls(results) + await test_cancel_while_blocked_reading(results) + await test_close_during_connect(results) + await test_close_prevents_reconnect_after_backoff(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())