fix(streaming): bound the ActionCable setup and repair the reconnect lifecycle (#108)

[karlwaldman]

Closes #108.

Reproduced first, against current main

Five PRs merged into this repo today, so the first step was confirming the
defects are live at 7982b0b rather than already fixed. A local probe with a
fake ActionCable socket (no production websocket, no network):

A: NO SETUP DEADLINE - blew past open_timeout=0.1s, still hung at 1.00s
A: socket closed after cancellation? False  <-- leak if False
B: raised ConnectionError: Subscription rejected; confirm the API k...
B: socket closed after failed __aenter__? False  <-- leak if False
C: raised OSError: connection refused after 1 reconnect attempt(s) of budget 10

Both defects in the issue reproduce exactly as described: the handshake is
unbounded, cancelling leaks the upgraded socket, and one failed reconnect
escapes a budget of ten.

The same probe after this change:

A: BOUNDED at 0.10s -> ActionCable setup timed out after 0.1s (connect, welcome, confirm_subscription). Raise setup_timeout if the server needs longer, or check the /cable endpoint.
A: socket closed after cancellation? True  <-- leak if False
B: raised StreamAuthError: Subscription rejected; confirm the API k...
B: socket closed after failed __aenter__? True  <-- leak if False
C: raised ConnectionError: Stream lost after 10 reconnect attempts after 10 reconnect attempt(s) of budget 10

What was wrong

open_timeout was passed to websockets.connect and governed the transport
upgrade only. The two awaits that follow it — _await_welcome and the
confirm_subscription wait inside _subscribe — had no deadline of any kind,
so a socket that upgrades and then goes silent hangs the caller forever, well
past the timeout they configured. Cancelling out of that hang left the upgraded
socket open, because self._ws was assigned before the handshake and nothing
closed it on the failure path. The same hole swallowed a rejected subscription:
__aenter__ raised, so __aexit__ never ran, so the socket leaked.

In _iterate, await self._reconnect() sat inside the except ConnectionClosed: block, so any exception it raised propagated out of the
iterator. A refused reconnect ended the stream on the first attempt regardless
of max_reconnect_attempts.

What changed

One bounded setup lifecycle. connect() now runs open + welcome +
confirm_subscription inside a single asyncio.wait_for. The deadline is a new
setup_timeout parameter on PriceStream and client.stream.prices(...),
which defaults to open_timeout — so the answer to "does open_timeout
cover protocol setup?" is now yes by default, and there is no unbounded wait on
either setting. open_timeout is still handed to the transport for the upgrade
itself. On expiry the caller gets a ConnectionError naming the timeout and the
three phases it covers.

Every allocated socket is closed. _setup puts the socket in a box that
connect() — the caller of wait_for, and therefore not the coroutine being
cancelled — can reach, and closes it in the failure path. That covers timeout,
subscription rejection, server disconnect, caller cancellation mid-setup, and
a failed __aenter__. Teardown itself is bounded by TEARDOWN_TIMEOUT so a
socket that will not close cannot wedge close(). _reconnect closes the
socket it is replacing, and an iterator that finishes releases its socket rather
than leaving it to garbage collection.

A real reconnect budget. _reconnect_with_budget consumes one attempt per
failure with the existing backoff, and raises the same Stream lost after N reconnect attempts error once the budget is spent. Transient failures are
OSError (which covers refused/reset connections and the bounded-setup
timeout), asyncio.TimeoutError, and websockets.WebSocketException.

Permanent failures stop immediately. A server disconnect or
reject_subscription now raises StreamAuthError, a subclass of
ConnectionError — existing except ConnectionError handlers are unaffected,
and test_rejected_subscription_raises passes unchanged — and the reconnect
loop re-raises it rather than retrying a rejected API key ten times. Exported
from oilpriceapi and oilpriceapi.streaming.

Close and cancel end the lifecycle. close() is idempotent, sets _closed
before anything else, sends the unsubscribe under a bounded timeout, and closes
the socket in a finally. connect() on a closed stream raises instead of
opening a new socket; closing during backoff stops the reconnect before it opens
one.

No new streaming framework, no background supervisor, no new task is ever
spawned by the client. The PriceStream / async with / async for
abstraction is unchanged, and the version stays at 1.14.0.

Sync counterpart

There is none — streaming is async-only (AsyncStreamNamespace is wired into
AsyncOilPriceAPI only; OilPriceAPI has no stream attribute). There was no
parallel sync path to change. test_streaming_is_async_only pins that, so if a
sync stream is ever added the test fails and forces the same lifecycle onto it.

Tests

17 new tests in tests/unit/test_streaming_lifecycle.py, written before the
fix. Every await in them carries its own deadline (DEADLINE = 2.0), so a
regression fails fast instead of blocking CI, and every case asserts that every
socket the fake connector handed out was closed.

Covered: upgraded socket with no welcome; welcome with no confirmation;
setup_timeout overriding open_timeout; rejected subscription; server
disconnect during the handshake; failed __aenter__; caller cancellation
mid-setup; a transient reconnect failure consuming the full budget; reconnect
succeeding within the budget; a permanent rejection on reconnect stopping at
once; reconnect closing the previous socket; close() preventing any further
connect; close() idempotence; closing during backoff; cancelling an active
stream leaving no pending task or socket; and async-only parity.

Red — new tests against the pre-fix sources

Produced by git checkout origin/main -- oilpriceapi/streaming/client.py oilpriceapi/streaming/__init__.py oilpriceapi/__init__.py and re-running, to
prove the tests are red-capable rather than tautological:

collected 17 items

tests/unit/test_streaming_lifecycle.py FFFFFFFFFFFFF.F..                 [100%]

=========================== short test summary info ============================
FAILED tests/unit/test_streaming_lifecycle.py::test_upgraded_socket_with_no_welcome_is_bounded
FAILED tests/unit/test_streaming_lifecycle.py::test_welcome_without_confirmation_is_bounded
FAILED tests/unit/test_streaming_lifecycle.py::test_setup_timeout_overrides_open_timeout
FAILED tests/unit/test_streaming_lifecycle.py::test_rejected_subscription_closes_the_socket
FAILED tests/unit/test_streaming_lifecycle.py::test_server_disconnect_during_handshake_closes_the_socket
FAILED tests/unit/test_streaming_lifecycle.py::test_failed_aenter_closes_the_socket
FAILED tests/unit/test_streaming_lifecycle.py::test_cancellation_mid_setup_closes_the_socket
FAILED tests/unit/test_streaming_lifecycle.py::test_transient_reconnect_failure_consumes_the_budget
FAILED tests/unit/test_streaming_lifecycle.py::test_reconnect_succeeds_within_the_budget
FAILED tests/unit/test_streaming_lifecycle.py::test_permanent_rejection_on_reconnect_stops_immediately
FAILED tests/unit/test_streaming_lifecycle.py::test_permanent_failure_is_a_distinguishable_error_type
FAILED tests/unit/test_streaming_lifecycle.py::test_reconnect_closes_the_previous_socket
FAILED tests/unit/test_streaming_lifecycle.py::test_close_prevents_any_further_connect
FAILED tests/unit/test_streaming_lifecycle.py::test_close_during_backoff_stops_reconnecting
========================= 14 failed, 3 passed in 4.37s =========================

Sample failure text from that run, with tracebacks on:

E   AssertionError: 2 of 2 allocated socket(s) left open (indices [0, 1])
E   AssertionError: the dropped socket was never closed
E   OSError: refused
E   ImportError: cannot import name 'StreamAuthError' from 'oilpriceapi.streaming.client'

The three that pass pre-fix (test_close_is_idempotent,
test_cancelling_an_active_stream_leaves_nothing_pending,
test_streaming_is_async_only) are regression guards, kept deliberately.

Green — new tests plus the existing streaming suite, unmodified

collected 30 items

tests/unit/test_streaming_lifecycle.py .................                 [ 56%]
tests/unit/test_streaming.py .............                               [100%]

============================== 30 passed in 0.52s ==============================

Full suite

Before, on clean main:

================== 3 failed, 864 passed, 63 skipped in 6.75s ===================

After:

================== 3 failed, 881 passed, 63 skipped in 5.85s ===================

881 = 864 + the 17 new tests; nothing else moved. The three failures are
tests/integration/test_demo_contract.py making live calls and getting HTTP 429
— environmental, identical before and after, unrelated to this change.

ruff check oilpriceapi/ and mypy oilpriceapi/ --ignore-missing-imports both
clean.

Compatibility and risk

Public API is additive: setup_timeout is a new keyword argument defaulting to
existing behaviour-preserving open_timeout, and StreamAuthError subclasses
ConnectionError. Two behaviour changes worth naming: a handshake that used to
hang forever now raises a ConnectionError at the configured timeout, and a
stream that has been closed raises on connect() instead of silently opening a
new socket. Both are the point of the fix. Rollback is a straight revert of this
commit.

Do not merge without review; this is one focused, reversible PR and nothing here
was published.

🤖 Generated with Claude Code

https://claude.ai/code/session_015ao5paex73xXvuM424Libo

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Advanced

Run ID: 3720658c-5ffa-4e13-ad18-eda9edd85220

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}