fix: harden SSE stream lifecycle (owned listen task, connect single-flight, real close) - #23
Conversation
…light, real close) Three lifecycle defects let concurrent connects overwrite the single response slot, left the response open after the last listener was removed, and let close() leave a running listener free to reconnect since it never stopped the task or survived cancellation. TeslemetryStream now owns exactly one listen task (auto-registration only starts it when absent/done; a second explicit listen() call joins the owner instead of racing it), connect() serializes the GET behind a lock and discards a response that arrives after a stop, and close() is a real stop: it flips active off, cancels the owned task, and closes the response, while internal reconnect paths use a new _close_response() that leaves active alone.
…oop on 3.9) Constructing asyncio.Lock() eagerly in __init__ broke Python 3.9, whose Lock still calls get_event_loop() at construction time - streams are commonly built before any event loop is running. Create it on first use inside connect() instead, which always runs with a loop active.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7e7cf6951b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| and not existing_task.done() | ||
| and existing_task is not current_task | ||
| ): | ||
| await existing_task |
There was a problem hiding this comment.
When a second listen() caller is cancelled—for example, because it is wrapped in asyncio.wait_for()—cancellation propagates through this bare await and cancels existing_task too. The owner's finally then closes the SSE response and clears _listen_task, while existing registered listeners prevent async_add_listener() from scheduling a replacement, leaving the stream silently stopped; await asyncio.shield(existing_task) so cancelling a joiner cannot cancel the owner.
AGENTS.md reference: AGENTS.md:L19-L19
Useful? React with 👍 / 👎.
1 participant
Intent
teslemetry_stream/stream.pyper a leak audit that proved three real defects, none of which are reachable through current stock Home Assistant's usage (one manual listener task, explicitclose()on unload) but are real bugs for any other caller:asyncio.create_task(self.listen()), and neitherlisten()norconnect()rejected a concurrent caller - two concurrent connects could both open a request and overwrite the single_responseslot, orphaning the first response.active = False- it didn't close the response or cancel the listener task, leaving an open response until timeout, session shutdown, or GC.close()closed the response but didn't stop anything:activestayed true so a running listener could reconnect, and with nofinallyaroundlisten(), task cancellation bypassed cleanup entirely and retained the response.TeslemetryStreamnow owns exactly one_listen_task. Auto-registration only starts it when absent/done, and a second explicitlisten()call joins the existing owner instead of racing it for the same connection.connect()serializes the GET behind a lock and re-checksactivebefore publishing the response, discarding one that arrives after a stop.close()is now a real stop - it flipsactiveoff, cancels the owned task, and closes the response - while a new internal_close_response()(response-only,activeuntouched) is what the ordinary EOF/error reconnect paths andlisten()'sfinallyuse, so cancellation always cleans up.close()/disconnect()/connect()/listen()keep their existing signatures and the single-listener happy path is unaffected. This is a bug-fix release - I'd call it a patch bump.tests/test_stream_lifecycle.pyadds the five regression scenarios from the audit: add/remove/re-add before the task runs, two concurrentlisten()calls, cancellation while blocked reading content, close during connect, and close preventing reconnect after backoff.