diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bb39b4a..2c199443 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,6 +93,16 @@ the raw override never did. - **`_decrypt_keys_for`** (`space_access.py`). No caller. Link-key selection stays inline at the access-resolution sites. +### `starfish-sdk` + +#### Changed +- **`subscribe_changes` retries URL-factory failures.** Resolving a callable + `url` now sits in the reconnect loop. A factory exception (for example a + token refresh) backs off and retries with the same cap as a transport error + (`httpx.HTTPError`, `OSError`). Exceptions from `on_status` propagate. + Backoff still resets when the connection opens (HTTP 2xx), not on the first + frame. The docstring no longer says the generator never raises. + ## 3.0.0-alpha.71 Adds device-code space-join pairing to `starfish-spaces` — the "requester diff --git a/packages/python/client/starfish_sdk/events.py b/packages/python/client/starfish_sdk/events.py index f2077d3d..2ffc120b 100644 --- a/packages/python/client/starfish_sdk/events.py +++ b/packages/python/client/starfish_sdk/events.py @@ -105,9 +105,15 @@ async def subscribe_changes( Yields: Parsed event data dicts from ``data:`` lines of each SSE frame. - The generator runs until it is closed by the caller (``aclose()`` or a - ``break`` inside an ``async for`` loop). It never raises — network errors - trigger a reconnect after the backoff delay. + Retried with capped exponential backoff: transport errors + (``httpx.HTTPError`` and ``OSError``) and any exception raised while + resolving a callable ``url``. Exceptions from ``on_status`` propagate to + the caller, as do any other errors once the URL has been resolved and the + failure is not a transport error. + + Backoff resets when a connection opens (HTTP 2xx), not when the first + frame arrives. The generator runs until the caller closes it + (``aclose()`` or ``break`` inside an ``async for`` loop). """ import httpx # local import so the module can be imported without httpx installed in tests that don't need it @@ -119,26 +125,42 @@ async def subscribe_changes( try: while True: - resolved_url = url() if callable(url) else url + # Factory failures are attempt failures: resolve inside the retry + # loop so a transient url() error backs off instead of ending the + # subscription. on_status errors are captured and re-raised below + # so they are not treated as transport failures. + try: + resolved_url = url() if callable(url) else url + except GeneratorExit: + return + except Exception: + await asyncio.sleep(backoff) + backoff = min(backoff * 2, max_backoff) + continue + + callback_error: Exception | None = None try: async with client.stream("GET", resolved_url, headers=_headers) as resp: if on_status is not None: - on_status(resp.status_code) - - # On non-2xx, back off and retry. - if resp.status_code < 200 or resp.status_code >= 300: - await asyncio.sleep(backoff) - backoff = min(backoff * 2, max_backoff) - continue - - # Successful connect — reset backoff. - backoff = initial_backoff_ms / 1000.0 - carry = "" - - async for chunk in resp.aiter_text(): - frames, carry = parse_sse_frames(chunk, carry) - for frame in frames: - yield frame + try: + on_status(resp.status_code) + except Exception as exc: + callback_error = exc + if callback_error is None: + # On non-2xx, back off and retry. + if resp.status_code < 200 or resp.status_code >= 300: + await asyncio.sleep(backoff) + backoff = min(backoff * 2, max_backoff) + continue + + # Successful connect — reset backoff. + backoff = initial_backoff_ms / 1000.0 + carry = "" + + async for chunk in resp.aiter_text(): + frames, carry = parse_sse_frames(chunk, carry) + for frame in frames: + yield frame except (httpx.HTTPError, OSError): # Network error — fall through to backoff and retry. @@ -146,6 +168,9 @@ async def subscribe_changes( except GeneratorExit: return + if callback_error is not None: + raise callback_error + await asyncio.sleep(backoff) backoff = min(backoff * 2, max_backoff) finally: diff --git a/packages/python/client/tests/test_events.py b/packages/python/client/tests/test_events.py index 1b1a99d8..b2d164e8 100644 --- a/packages/python/client/tests/test_events.py +++ b/packages/python/client/tests/test_events.py @@ -1,8 +1,55 @@ -"""Tests for starfish_sdk.events — SSE frame parser.""" +"""Tests for starfish_sdk.events — SSE frame parser and subscribe_changes.""" +import asyncio +import gc + +import httpx import pytest -from starfish_sdk.events import parse_sse_frames +from starfish_sdk.events import parse_sse_frames, subscribe_changes + +_FRAME = 'data: {"n":1}\n\n' +_FRAME_2 = 'data: {"n":2}\n\n' +_URL = "https://events.test/stream" + + +def _patch_sleep(monkeypatch) -> list[int]: + delays: list[int] = [] + + async def fake_sleep(seconds: float) -> None: + delays.append(int(round(seconds * 1000))) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + return delays + + +def _scripted_client(steps: list) -> tuple[httpx.AsyncClient, list[str]]: + """MockTransport client that plays `steps` (Response or exception) in order.""" + index = {"i": 0} + urls: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + urls.append(str(request.url)) + step = steps[index["i"]] + index["i"] += 1 + if isinstance(step, Exception): + raise step + return step + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + return client, urls + + +def _ok(body: str = _FRAME) -> httpx.Response: + return httpx.Response(200, text=body) + + +def _status(code: int) -> httpx.Response: + return httpx.Response(code, text="") + + +def _connect_error(message: str) -> httpx.ConnectError: + return httpx.ConnectError(message) # ── parse_sse_frames ────────────────────────────────────────────────────────── @@ -109,3 +156,233 @@ def test_carry_preserved_across_comment_and_data(): frames2, carry2 = parse_sse_frames('data: {"z":9}\n\n', carry1) assert frames2 == [{"z": 9}] assert carry2 == "" + + +# ── subscribe_changes ───────────────────────────────────────────────────────── + + +async def _first_frame(**kwargs) -> dict: + kwargs.setdefault("initial_backoff_ms", 10) + kwargs.setdefault("max_backoff_ms", 40) + gen = subscribe_changes(**kwargs) + frame = await gen.__anext__() + await gen.aclose() + return frame + + +async def test_subscribe_reconnects_after_connect_error(monkeypatch): + # T1: two network errors, then one frame. on_status sees only the 200. + _patch_sleep(monkeypatch) + statuses: list[int] = [] + client, _urls = _scripted_client([ + _connect_error("down"), + _connect_error("down"), + _ok(), + ]) + frame = await _first_frame( + url=_URL, + session=client, + on_status=statuses.append, + ) + assert frame == {"n": 1} + assert statuses == [200] + await client.aclose() + + +async def test_subscribe_backoff_doubles_and_caps(monkeypatch): + # T2: five failures then a frame. Delays double and stick at the cap. + delays = _patch_sleep(monkeypatch) + client, _urls = _scripted_client([_connect_error("down")] * 5 + [_ok()]) + frame = await _first_frame(url=_URL, session=client) + assert frame == {"n": 1} + assert delays == [10, 20, 40, 40, 40] + await client.aclose() + + +async def test_subscribe_backoff_resets_after_successful_connect(monkeypatch): + # T3: errors at 10 then 20, a clean 200, then the next delay is the reset 10. + delays = _patch_sleep(monkeypatch) + client, _urls = _scripted_client([ + _connect_error("down"), + _connect_error("down"), + _ok(), + _connect_error("down"), + _ok(_FRAME_2), + ]) + gen = subscribe_changes( + url=_URL, + session=client, + initial_backoff_ms=10, + max_backoff_ms=40, + ) + frames = [] + async for frame in gen: + frames.append(frame) + if len(frames) == 2: + break + assert frames == [{"n": 1}, {"n": 2}] + assert delays[:3] == [10, 20, 10] + await client.aclose() + + +async def test_subscribe_retries_non_2xx_without_yielding(monkeypatch): + # T4: 503 and 500 are not yielded. Only the following 200 frame is. + delays = _patch_sleep(monkeypatch) + client, _urls = _scripted_client([_status(503), _status(500), _ok()]) + frame = await _first_frame(url=_URL, session=client) + assert frame == {"n": 1} + assert delays == [10, 20] + await client.aclose() + + +async def test_subscribe_on_status_called_before_backoff(monkeypatch): + # T5: on_status runs for every attempt, before that attempt's backoff sleep. + log: list[tuple] = [] + + async def fake_sleep(seconds: float) -> None: + log.append(("sleep", int(round(seconds * 1000)))) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + client, _urls = _scripted_client([_status(503), _status(500), _ok()]) + + def on_status(code: int) -> None: + log.append(("status", code)) + + frame = await _first_frame(url=_URL, session=client, on_status=on_status) + assert frame == {"n": 1} + assert log == [ + ("status", 503), + ("sleep", 10), + ("status", 500), + ("sleep", 20), + ("status", 200), + ] + await client.aclose() + + +async def test_subscribe_reresolves_callable_url(monkeypatch): + # T6: the URL factory is called again on every attempt. + _patch_sleep(monkeypatch) + planned = [ + "https://events.test/u1", + "https://events.test/u2", + "https://events.test/u3", + ] + produced: list[str] = [] + + def factory() -> str: + nxt = planned[len(produced)] + produced.append(nxt) + return nxt + + client, requested = _scripted_client([ + _connect_error("down"), + _connect_error("down"), + _ok(), + ]) + frame = await _first_frame(url=factory, session=client) + assert frame == {"n": 1} + assert requested == planned + await client.aclose() + + +def _patch_owned_client(monkeypatch) -> list[httpx.AsyncClient]: + created: list[httpx.AsyncClient] = [] + real = httpx.AsyncClient + + def factory(*args, **kwargs): + kwargs["transport"] = httpx.MockTransport(lambda request: _ok()) + client = real(*args, **kwargs) + created.append(client) + return client + + monkeypatch.setattr(httpx, "AsyncClient", factory) + return created + + +async def test_subscribe_closes_owned_client_on_aclose(monkeypatch): + # T7: with no caller session, aclose() closes the client this call opened. + created = _patch_owned_client(monkeypatch) + gen = subscribe_changes(url=_URL, initial_backoff_ms=10, max_backoff_ms=40) + frame = await gen.__anext__() + assert frame == {"n": 1} + await gen.aclose() + assert len(created) == 1 + assert created[0].is_closed + + +async def test_subscribe_leaves_caller_session_open(): + # T8: a session passed by the caller stays open after the generator closes. + client, _urls = _scripted_client([_ok()]) + gen = subscribe_changes( + url=_URL, + session=client, + initial_backoff_ms=10, + max_backoff_ms=40, + ) + frame = await gen.__anext__() + assert frame == {"n": 1} + await gen.aclose() + assert client.is_closed is False + await client.aclose() + + +async def test_subscribe_break_closes_owned_client(monkeypatch): + # T9: breaking out of async for closes the owned client. + # CPython 3.12 does not aclose the generator inside the loop frame. The + # close is scheduled once that frame is collected, so leave the loop in a + # nested coroutine and wait for the finalizer task. + created = _patch_owned_client(monkeypatch) + + async def consume() -> None: + async for frame in subscribe_changes( + url=_URL, + initial_backoff_ms=10, + max_backoff_ms=40, + ): + assert frame == {"n": 1} + break + + await consume() + gc.collect() + await asyncio.sleep(0) + pending = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()] + if pending: + await asyncio.gather(*pending) + assert len(created) == 1 + assert created[0].is_closed + + +async def test_subscribe_retries_url_factory_and_propagates_on_status(monkeypatch): + # T10 (option C): a URL-factory RuntimeError is retried; on_status raises. + delays = _patch_sleep(monkeypatch) + calls = {"n": 0} + + def factory() -> str: + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("token refresh failed") + return _URL + + client, _urls = _scripted_client([_ok()]) + frame = await _first_frame(url=factory, session=client) + assert frame == {"n": 1} + assert calls["n"] == 2 + assert delays == [10] + + def on_status(_code: int) -> None: + raise RuntimeError("on_status bug") + + client2, _urls = _scripted_client([_ok()]) + gen = subscribe_changes( + url=_URL, + session=client2, + on_status=on_status, + initial_backoff_ms=10, + max_backoff_ms=40, + ) + with pytest.raises(RuntimeError, match="on_status bug"): + await gen.__anext__() + assert delays == [10] + await client.aclose() + await client2.aclose()