diff --git a/docs/realtime.md b/docs/realtime.md index 754cccb8..c05fd4b1 100644 --- a/docs/realtime.md +++ b/docs/realtime.md @@ -42,6 +42,25 @@ Channel names receive their type prefix, so `updates` becomes `broadcast:updates Use the same event loop for a client's realtime operations. The following channel examples belong inside an async function with an authenticated `client`. +## Connection failures + +An auth-only public key cannot connect to realtime. Include `realtime.connect` +and `realtime.subscribe` on the key. A terminal disconnect during startup rejects +the pending subscription promptly with `RealtimeConnectionError`, preserving the +server's reason and optional numeric code. Failed and cancelled starts release +their transport and session binding so another attempt can use fresh credentials. +Connection startup is bounded to ten seconds; retryable interruptions can still +recover within that deadline. + +```python +from volcano_sdk import RealtimeConnectionError + +try: + await channel.subscribe() +except RealtimeConnectionError as error: + print(str(error), error.code) # code: int | None +``` + ## Pause or remove channels ```python diff --git a/src/volcano_sdk/__init__.py b/src/volcano_sdk/__init__.py index 79596c0e..2e0357d8 100644 --- a/src/volcano_sdk/__init__.py +++ b/src/volcano_sdk/__init__.py @@ -9,6 +9,7 @@ NotFoundError, PermissionDeniedError, RateLimitedError, + RealtimeConnectionError, ServerError, SessionChangedError, TransportError, @@ -104,6 +105,7 @@ "PostgresChange", "RateLimitedError", "RealtimeConnectContext", + "RealtimeConnectionError", "RealtimeDisconnectContext", "RealtimeErrorContext", "RealtimePresenceInfo", diff --git a/src/volcano_sdk/_realtime_connection.py b/src/volcano_sdk/_realtime_connection.py index 9ed25e96..62fa00f6 100644 --- a/src/volcano_sdk/_realtime_connection.py +++ b/src/volcano_sdk/_realtime_connection.py @@ -4,6 +4,7 @@ import asyncio import inspect +from contextlib import suppress from dataclasses import dataclass from itertools import count from typing import ( @@ -63,6 +64,7 @@ invoke_async, response_payload, ) +from .errors import RealtimeConnectionError if TYPE_CHECKING: from collections.abc import Callable, Iterator, Mapping @@ -89,9 +91,13 @@ class ManagedChannel(Generic[FacadeT]): class ClientEvents(ClientEventHandler): def __init__(self, enqueue: Callable[[ConnectionContext], None]) -> None: self.enqueue: Callable[[ConnectionContext], None] = enqueue + self.started: asyncio.Event = asyncio.Event() + self.failure: RealtimeConnectionError | None = None + self.is_terminal: Callable[[], bool] = lambda: False @override async def on_connected(self, ctx: object) -> None: + self.started.set() client = _native.native_attribute(ctx, "client") self.enqueue( RealtimeConnectContext(client=client if isinstance(client, str) else None), @@ -101,6 +107,14 @@ async def on_connected(self, ctx: object) -> None: async def on_disconnected(self, ctx: object) -> None: code = _native.native_attribute(ctx, "code") reason = _native.native_attribute(ctx, "reason") + if self.is_terminal(): + self.failure = RealtimeConnectionError( + reason + if isinstance(reason, str) and reason + else "Connection disconnected", + code=code if isinstance(code, int) else None, + ) + self.started.set() self.enqueue( RealtimeDisconnectContext( code=code if isinstance(code, int) else None, @@ -108,6 +122,12 @@ async def on_disconnected(self, ctx: object) -> None: ), ) + async def wait_ready(self, connection: _native.VolcanoCentrifugeConnection) -> None: + if not connection.is_connected: + _ = await self.started.wait() + if self.failure is not None: + raise self.failure + @override async def on_error(self, ctx: object) -> None: code = _native.native_attribute(ctx, "code") @@ -489,21 +509,26 @@ async def _connect_locked(self) -> _native.VolcanoCentrifugeConnection: _generation, lineage, session = self.client_context.capture_session_binding() if session is None: raise RuntimeError(NO_ACTIVE_SESSION) + events = ClientEvents(self.enqueue_connection_callbacks) connection = _native.VolcanoCentrifugeConnection( self.client_factory( self._address(), - events=ClientEvents(self.enqueue_connection_callbacks), + events=events, token=session.access_token, get_token=self._token, ) ) + events.is_terminal = lambda: connection.is_disconnected self.connection_session_lineage = lineage self.connection_access_token = session.access_token try: - await connection.connect() + async with asyncio.timeout(10): + await connection.connect() + await events.wait_ready(connection) except BaseException: self.connection_session_lineage = None self.connection_access_token = None + await self._discard_failed_connection(connection) raise try: current_session = self._session_for_lineage(lineage) @@ -518,6 +543,14 @@ async def _connect_locked(self) -> _native.VolcanoCentrifugeConnection: self.connection = connection return connection + @staticmethod + async def _discard_failed_connection( + connection: _native.VolcanoCentrifugeConnection, + ) -> None: + # Cleanup must not replace the original connection failure/cancellation. + with suppress(Exception, asyncio.CancelledError): + await connection.disconnect() + async def subscribe(self, channel: ChannelState) -> None: # A later stop supersedes this request, including time spent waiting for locks. generation = channel.subscribe_generation diff --git a/src/volcano_sdk/_realtime_transport.py b/src/volcano_sdk/_realtime_transport.py index 3493ed94..b978168e 100644 --- a/src/volcano_sdk/_realtime_transport.py +++ b/src/volcano_sdk/_realtime_transport.py @@ -261,6 +261,10 @@ def is_connected(self) -> bool: state = self._connection.state return native_attribute(state, "value") == "connected" + @property + def is_disconnected(self) -> bool: + return native_attribute(self._connection.state, "value") == "disconnected" + def new_subscription( self, name: str, diff --git a/src/volcano_sdk/_tests/test_realtime.py b/src/volcano_sdk/_tests/test_realtime.py index 272b38ac..90415f33 100644 --- a/src/volcano_sdk/_tests/test_realtime.py +++ b/src/volcano_sdk/_tests/test_realtime.py @@ -564,7 +564,7 @@ def __init__(self, events: object = None) -> None: self.presence_entered: asyncio.Event | None = None self.presence_release: asyncio.Event | None = None self.disconnect_probe: Callable[[], None] | None = None - self.disconnect_error: Exception | None = None + self.disconnect_error: BaseException | None = None self.subscription: FakeSubscription | None = None self._subs: MutableMapping[str, FakeSubscription] = {} @@ -3650,6 +3650,7 @@ async def connect() -> None: entered.set() _ = await asyncio.wait_for(release.wait(), timeout=2) official.calls.append("connect") + official.state = FakeConnectionState(value="connected") monkeypatch.setattr(official, "connect", connect) @@ -3783,6 +3784,7 @@ async def connect() -> None: entered.set() _ = await asyncio.wait_for(release.wait(), timeout=2) official.calls.append("connect") + official.state = FakeConnectionState(value="connected") monkeypatch.setattr(official, "connect", connect) client = VolcanoClient( diff --git a/src/volcano_sdk/_tests/test_realtime_connection_boundaries.py b/src/volcano_sdk/_tests/test_realtime_connection_boundaries.py index e46eb9f6..496a5199 100644 --- a/src/volcano_sdk/_tests/test_realtime_connection_boundaries.py +++ b/src/volcano_sdk/_tests/test_realtime_connection_boundaries.py @@ -9,6 +9,7 @@ from volcano_sdk import ( RealtimeConnectContext, + RealtimeConnectionError, RealtimeDisconnectContext, RealtimeErrorContext, Session, @@ -32,7 +33,12 @@ from .client_inspection import InspectedClient from .realtime_probes import channel_state, completed_operation, realtime_state -from .test_realtime import FakeCentrifugeClient, FakeCentrifugeFactory, FakeSubscription +from .test_realtime import ( + FakeCentrifugeClient, + FakeCentrifugeFactory, + FakeConnectionState, + FakeSubscription, +) if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -70,6 +76,7 @@ async def test_failed_connect_clears_credentials_before_retry( await channel.subscribe() assert caught.value is failure + assert native.calls == ["disconnect"] assert realtime_state(client.realtime).connection is None assert realtime_state(client.realtime).connection_session_lineage is None assert realtime_state(client.realtime).connection_access_token is None @@ -84,6 +91,112 @@ async def test_failed_connect_clears_credentials_before_retry( await client.realtime.disconnect() +class RejectedConnection(FakeCentrifugeClient): + @override + async def connect(self) -> None: + assert self.events is not None + await self.events.on_disconnected( + SimpleNamespace(code=3507, reason="permission denied") + ) + + +@pytest.mark.parametrize( + "cleanup_failure", + [None, RuntimeError("cleanup failed"), asyncio.CancelledError("cleanup cancelled")], +) +async def test_terminal_connection_failure_preserves_reason_and_stops_transport( + cleanup_failure: BaseException | None, +) -> None: + native = RejectedConnection() + native.disconnect_error = cleanup_failure + client = VolcanoClient( + anon_key="anon", + access_token="access", + _realtime_client_factory=FakeCentrifugeFactory(native), + ) + try: + with pytest.raises( + RealtimeConnectionError, match="permission denied" + ) as caught: + await asyncio.wait_for(client.realtime.channel("room").subscribe(), 0.2) + assert caught.value.code == 3507 + assert native.calls == ["disconnect"] + assert realtime_state(client.realtime).connection is None + finally: + await client.realtime.disconnect() + + +async def test_unbound_connection_events_do_not_treat_a_disconnect_as_terminal() -> ( + None +): + events = ClientEvents(lambda _context: None) + await events.on_disconnected(SimpleNamespace(code=3001, reason="transport closed")) + + assert events.failure is None + assert not events.started.is_set() + + +async def test_connect_waits_for_a_delayed_native_connection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + native = FakeCentrifugeClient() + entered = asyncio.Event() + + async def connect() -> None: + native.state = FakeConnectionState(value="connecting") + entered.set() + await completed_operation(None) + + monkeypatch.setattr(native, "connect", connect) + client = VolcanoClient( + anon_key="anon", + access_token="access", + _realtime_client_factory=FakeCentrifugeFactory(native), + ) + connecting = asyncio.create_task(realtime_state(client.realtime).connect()) + try: + _ = await asyncio.wait_for(entered.wait(), timeout=0.2) + assert not connecting.done() + assert not client.realtime.is_connected + native.state = FakeConnectionState(value="connected") + assert native.events is not None + await native.events.on_connected(SimpleNamespace(client="client-123")) + _ = await asyncio.wait_for(connecting, timeout=0.2) + assert client.realtime.is_connected + finally: + _ = connecting.cancel() + _ = await asyncio.gather(connecting, return_exceptions=True) + await client.realtime.disconnect() + + +async def test_connection_startup_times_out_within_its_ten_second_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + native = FakeCentrifugeClient() + + async def connect() -> None: + _ = await asyncio.Event().wait() + + monkeypatch.setattr(native, "connect", connect) + client = VolcanoClient( + anon_key="anon", + access_token="access", + _realtime_client_factory=FakeCentrifugeFactory(native), + ) + try: + # The outer deadline distinguishes the SDK's ten-second timeout from + # an absent deadline or one extended beyond its documented bound. + async with asyncio.timeout(10.75): + with pytest.raises(TimeoutError): + _ = await realtime_state(client.realtime).connect() + assert native.calls == ["disconnect"] + assert realtime_state(client.realtime).connection_session_lineage is None + assert realtime_state(client.realtime).connection_access_token is None + assert not client.realtime.is_connected + finally: + await client.realtime.disconnect() + + async def test_connection_requires_a_session_before_constructing_transport() -> None: native = FakeCentrifugeClient() client = VolcanoClient( diff --git a/src/volcano_sdk/_tests/typing/realtime_subscriptions.py b/src/volcano_sdk/_tests/typing/realtime_subscriptions.py index d8bea6cf..f7b6a2a9 100644 --- a/src/volcano_sdk/_tests/typing/realtime_subscriptions.py +++ b/src/volcano_sdk/_tests/typing/realtime_subscriptions.py @@ -3,6 +3,7 @@ from collections.abc import Mapping from typing import assert_type +from volcano_sdk import RealtimeConnectionError from volcano_sdk._realtime_transport import ( ProjectAwareSubscriptions, ) @@ -52,3 +53,9 @@ async def channel_facade_types(realtime: Realtime) -> None: await realtime.remove_channel("members", channel_type="presence") await realtime.remove_all_channels() await realtime.disconnect() + + +def disconnect_code(error: BaseException) -> None: + """An exception handler can read numeric disconnect codes without casts.""" + if isinstance(error, RealtimeConnectionError): + _code = assert_type(error.code, int | None) diff --git a/src/volcano_sdk/errors.py b/src/volcano_sdk/errors.py index 3ca6184f..0d6eb703 100644 --- a/src/volcano_sdk/errors.py +++ b/src/volcano_sdk/errors.py @@ -67,3 +67,17 @@ class ServerError(VolcanoError): class TransportError(VolcanoError): """The request failed before receiving an API response.""" + + +class RealtimeConnectionError(RuntimeError): + """A terminal realtime disconnect before a connection becomes ready.""" + + def __init__(self, message: str, *, code: int | None = None) -> None: + """Preserve the disconnect reason and optional numeric transport code.""" + super().__init__(message) + self._code: int | None = code + + @property + def code(self) -> int | None: + """The transport's disconnect code, when available.""" + return self._code