Retry subscribe_changes URL factory failures

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

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,33 +125,52 @@ 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.
                 pass
             except GeneratorExit:
                 return
 
+            if callback_error is not None:
+                raise callback_error
+
             await asyncio.sleep(backoff)
             backoff = min(backoff * 2, max_backoff)
     finally:

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()