fix(realtime): surface terminal startup rejection reasons

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

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",

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,13 +107,27 @@ 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,
                 reason=reason if isinstance(reason, str) else 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

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,

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(

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(

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)

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