fix(supabase)!: share one lazily connecting realtime client per AsyncClient

MIGRATION.md

@@ -7,9 +7,9 @@ Abbreviated and contracted public names are spelled out. Wire formats do not cha
 | Package              | Before                                                           | After                                                                                 |
 | -------------------- | ---------------------------------------------------------------- | ------------------------------------------------------------------------------------- |
 | `realtime`           | `connect_once(hb_interval=..., hb_timeout=...)`                  | `connect_once(heartbeat_interval=..., heartbeat_timeout=...)`                         |
-| `realtime`           | `RealtimeClient(ws_connection, hb_interval=..., hb_timeout=...)` | `RealtimeClient(websocket_connection, heartbeat_interval=..., heartbeat_timeout=...)` |
+| `realtime`           | `RealtimeClient(ws_connection, hb_interval=..., hb_timeout=...)` | `RealtimeClient(url, token_callback, heartbeat_interval=..., heartbeat_timeout=...)`  |
 | `realtime`           | `RealtimeClient.hb_interval`                                     | `RealtimeClient.heartbeat_interval`                                                   |
-| `realtime`           | `connect_once.ws`                                                | `connect_once.websocket`                                                              |
+| `realtime`           | `connect_once.ws`                                                | Removed, `connect_once.client` holds the connected `RealtimeClient`                   |
 | `realtime`           | `RealtimeClient.channel(topic, params=...)`                      | `RealtimeClient.channel(topic, options=...)`                                          |
 | `realtime`           | `RealtimeChannel(socket, topic, params=...)`                     | `RealtimeChannel(socket, topic, options=...)`                                         |
 | `realtime`           | `RealtimeChannel.params`                                         | `RealtimeChannel.options`                                                             |
@@ -200,6 +200,18 @@ Every filter on the table counts, including `match()`, `in_()`, `or_()` and a ne
 
 `heartbeat_interval` and `heartbeat_timeout` are typed `float` and are in seconds. `connect_once`, `RealtimeClient` and `RealtimeClientOptions` raise `ValueError` when either one is not positive.
 
+## Realtime clients connect on the first subscribe
+
+`RealtimeClient` takes the realtime URL and a token callback instead of an open connection. It connects when the first channel subscribes, or when `connect()` is called, and every channel of the client shares that connection. The connection is closed when the last channel is unsubscribed or removed, so a client with no channels holds no connection. `aclose()` closes it too and leaves the channels unsubscribed. Subscribing again opens a new connection, which also replaces a connection that was lost.
+
+Messages queued on a channel, like the close message of `unsubscribe()`, can still be read from `channel.messages` after the connection is closed. Reading past them raises `NotConnectedError` instead of waiting.
+
+`AsyncClient.realtime` returns the same `RealtimeClient` every time, so channels made through it are visible to each other. It is closed by `AsyncClient.aclose()` and when the `AsyncClient` exits. Before, every read of `AsyncClient.realtime` returned a new `connect_once` that opened its own connection.
+
+`connect_once` keeps the client it connects in `connect_once.client`. The `websocket`, `listen_task` and `heartbeat_task` attributes are removed.
+
+`unsubscribe()` on a channel that is not joined, for example after `aclose()`, removes the channel without sending a leave message.
+
 ## Realtime `close()` is `aclose()`
 
 `RealtimeClient.close()` is renamed to `aclose()`, the name Python uses for an async close, so `contextlib.aclosing(client)` works. `close()` is removed.

sdk-compliance.yaml

@@ -861,14 +861,13 @@ features:
 
   realtime.client.connect:
     status: implemented
-    note: "connect_once(url, ...) is awaitable and an async context manager that opens the connection and returns a RealtimeClient; automatically_reconnect wraps it with reconnection."
+    note: "RealtimeClient connects when the first channel subscribes, or when connect() is called, and reuses that connection for every channel until the last channel is removed. AsyncClient.realtime is one RealtimeClient per supabase client, closed when the supabase client exits. connect_once(url, ...) is awaitable and an async context manager that connects a RealtimeClient and closes it on exit; automatically_reconnect wraps it with reconnection."
     symbols:
+      - RealtimeClient.connect
       - connect_once
       - automatically_reconnect
     supporting_symbols:
       - RealtimeClient
-      - RealtimeClient.listen
-      - RealtimeClient.heartbeat
       - RealtimeClient.endpoint_url
       - normalize_url
       - MaxRetriesExceeded

src/realtime/README.md

@@ -80,6 +80,7 @@ asyncio.run(main())
 - `REALTIME_URL` is `http://localhost:54321/realtime/v1` when developing locally with the Supabase CLI and `https://<project_ref>.supabase.co/realtime/v1` when connecting to your Supabase project. The client switches to `ws`/`wss` and appends `/websocket` itself.
 - The token callback is awaited when connecting and on every heartbeat, so it can return a refreshed user access token. It must return a JWT whose claims contain `exp` and `role` (an existing database role), or the publishable key.
 - `async with client.channel(...)` subscribes on entry and unsubscribes on exit. `subscribe()` raises if the server rejects the join.
+- `RealtimeClient(url, get_token)` creates a client without connecting. It connects when the first channel subscribes, shares that connection with every channel, and closes it when the last channel is unsubscribed or `aclose()` is called.
 - Channel name can be any `string`.
 - Use `realtime.client.automatically_reconnect(url, get_token, handler)` instead of `connect_once` to reconnect with exponential backoff when the connection drops.
 

src/realtime/src/realtime/channel.py

@@ -200,6 +200,7 @@ def __init__(
         self.presence: RealtimePresence = RealtimePresence()
         self.state = ChannelStates.CLOSED
         self.join_ref: str | None = None
+        self._subscribe_attempt: object | None = None
         self.message_stream: asyncio.Queue[Message] = asyncio.Queue()
         self.broadcast_endpoint_url = self._broadcast_endpoint_url()
 
@@ -216,11 +217,20 @@ async def __aexit__(self, *args: object) -> None:
             await self.unsubscribe()
 
     async def subscribe(self) -> ReplyMessage:
-        owner = self.socket.channels.setdefault(self.topic, self)
-        if owner is not self:
-            raise ValueError(
-                f"another channel is already registered for {self.topic!r}"
-            )
+        """
+        Join the channel, connecting the client first if it is not connected.
+
+        :raises RealtimeError: If the server rejects the join, or if
+            `unsubscribe()` or `RealtimeClient.aclose()` is called before the
+            join completes.
+        """
+        attempt = object()
+        self._subscribe_attempt = attempt
+        self._raise_if_topic_is_taken()
+        await self.socket.connect()
+        self._raise_if_join_was_stopped(attempt)
+        self._raise_if_topic_is_taken()
+        self.socket.channels[self.topic] = self
         while not self.message_stream.empty():
             self.message_stream.get_nowait()
         options = self.options
@@ -235,21 +245,67 @@ async def subscribe(self) -> ReplyMessage:
             ref=join_ref,
             join_ref=join_ref,
         )
-        msg = await self.socket._send(message)
+        self.join_ref = join_ref
+        self.state = ChannelStates.JOINING
+        failure = self.socket._connection_failure
+        try:
+            msg = await self.socket._send(message)
+        except asyncio.CancelledError:
+            self.state = ChannelStates.CLOSED
+            if self._subscribe_attempt is attempt:
+                self._subscribe_attempt = None
+                await self._leave_cancelled_join(join_ref, failure)
+            raise
+        except Exception:
+            self.state = ChannelStates.CLOSED
+            self._raise_if_join_was_stopped(attempt)
+            raise
         logger.info("Subscribe reply: %r", msg)
+        self._raise_if_join_was_stopped(attempt)
         if msg.payload.status != "ok":
+            self.state = ChannelStates.CLOSED
             raise RealtimeError(
                 f"error while subscribing to channel: {msg.payload.response!r}"
             )
-        self.join_ref = join_ref
         self.state = ChannelStates.JOINED
         return msg
 
+    def _raise_if_join_was_stopped(self, attempt: object) -> None:
+        if self._subscribe_attempt is not attempt:
+            raise RealtimeError(
+                f"the join of channel {self.topic!r} was stopped by unsubscribe() "
+                "or a closed connection before it completed"
+            )
+
+    async def _leave_cancelled_join(
+        self, join_ref: str, failure: asyncio.Future[Message] | None
+    ) -> None:
+        if self.socket._connection_failure is not failure:
+            return
+        leave = Message(
+            topic=self.topic,
+            event=ChannelEvents.leave,
+            ref=self.socket._make_ref(),
+            join_ref=join_ref,
+            payload={},
+        )
+        with contextlib.suppress(Exception):
+            await self.socket._send_no_wait(leave)
+
+    def _raise_if_topic_is_taken(self) -> None:
+        if self.socket.channels.get(self.topic, self) is not self:
+            raise ValueError(
+                f"another channel is already registered for {self.topic!r}"
+            )
+
     @property
     async def messages(self) -> AsyncIterator[ChannelMessage]:
         while True:
-            get_task = asyncio.create_task(self.message_stream.get())
-            message = await self.socket._get_message_or_raise(get_task)
+            if self.message_stream.empty():
+                get_task = asyncio.create_task(self.message_stream.get())
+                message = await self.socket._get_message_or_raise(get_task)
+            else:
+                message = self.message_stream.get_nowait()
             server_message = ServerMessageAdapter.validate_python(message)
             if isinstance(server_message, PresenceStateMessage):
                 for presence_event in self.presence._on_state_event(
@@ -265,10 +321,13 @@ async def messages(self) -> AsyncIterator[ChannelMessage]:
                 yield server_message
 
     async def unsubscribe(self) -> None:
+        self._subscribe_attempt = None
+        needs_leave = self.state in (ChannelStates.JOINED, ChannelStates.JOINING)
         self.state = ChannelStates.LEAVING
         self.socket._remove_channel(self)
         try:
-            await self._send_event(ChannelEvents.leave, payload={})
+            if needs_leave:
+                await self._send_event(ChannelEvents.leave, payload={})
         finally:
             self.state = ChannelStates.CLOSED
             self.message_stream.put_nowait(
@@ -280,6 +339,7 @@ async def unsubscribe(self) -> None:
                     join_ref=self.join_ref,
                 )
             )
+            await self.socket._close_if_unused()
 
     # Presence methods
     async def track(self, user_status: dict[str, Any]) -> ReplyMessage: