feat(realtime)!: move to protocol 2.0.0 with binary broadcast

sdk-compliance.yaml

@@ -1045,7 +1045,13 @@ features:
       - connect_once
   realtime.configuration.deferred_disconnect: not_implemented
   realtime.configuration.custom_logger: not_applicable
-  realtime.configuration.binary_protocol: not_implemented
+  realtime.configuration.binary_protocol:
+    status: implemented
+    note: "The client always speaks protocol 2.0.0. send_broadcast() sends a bytes payload as a binary frame, and binary broadcasts are received as bytes in BroadcastPayload.payload."
+    symbols:
+      - RealtimeChannel.send_broadcast
+    supporting_symbols:
+      - BroadcastPayload
   realtime.configuration.foreground_liveness_probe:
     status: not_applicable
     note: "Python has no app foreground lifecycle to hook into. Dropped connections are detected by the heartbeat timeout."

src/realtime/README.md

@@ -103,6 +103,7 @@ async with client.channel("broadcast-test", options) as channel:
 
 - With `ack=True`, `send_broadcast()` waits for the server to acknowledge the message and returns its reply.
 - With `listen_self=True`, the client receives the broadcast messages it sends.
+- A `bytes` payload is sent as a binary WebSocket frame, without JSON encoding, and receivers get it back as `bytes` in `message.payload.payload`.
 - `RealtimeChannelOptions().private(True)` makes the server use RLS to decide whether the user may join the channel.
 - Setting `replication_ready` to `true` instructs the server to emit a `system` message once the Postgres replication connection backing the channel is established and ready to stream changes. Its payload's `status` is `"ok"` (`message: "Replication connection established"`) on success or `"error"` if the connection is not ready in time.
 

src/realtime/src/realtime/_serializer.py

@@ -0,0 +1,118 @@
+"""Encode and decode the frames of the Realtime protocol 2.0.0.
+
+Text frames hold the JSON array ``[join_ref, ref, topic, event, payload]``.
+A broadcast whose payload is ``bytes`` is sent as a binary ``user broadcast
+push`` frame, and the server sends broadcasts to the client as binary
+``user broadcast`` frames, with either a JSON or a binary payload.
+"""
+
+import json
+from typing import TYPE_CHECKING, Any
+
+from pydantic import TypeAdapter
+
+from .message import Message
+from .types import ChannelEvents
+
+if TYPE_CHECKING:
+    from supabase_utils.types import JSON
+
+_USER_BROADCAST_PUSH = 3
+_USER_BROADCAST = 4
+_BINARY_ENCODING = 0
+_JSON_ENCODING = 1
+_USER_BROADCAST_HEADER_LENGTH = 5
+_MAX_FIELD_LENGTH = 255
+
+type _TextFrame = tuple[str | None, str | None, str, str, Any]
+_TextFrameAdapter: TypeAdapter[_TextFrame] = TypeAdapter(_TextFrame)
+
+
+def encode(message: Message) -> str | bytes:
+    payload = message.payload
+    user_event = payload.get("event")
+    user_payload = payload.get("payload")
+    if (
+        message.event == ChannelEvents.broadcast
+        and isinstance(user_event, str)
+        and isinstance(user_payload, bytes)
+    ):
+        return _encode_user_broadcast_push(message, user_event, user_payload)
+    frame = (message.join_ref, message.ref, message.topic, message.event, payload)
+    return _TextFrameAdapter.dump_json(frame).decode()
+
+
+def decode(frame: str | bytes) -> Message:
+    """Decode a frame, raising `ValueError` when it is malformed."""
+    if isinstance(frame, str):
+        join_ref, ref, topic, event, payload = _TextFrameAdapter.validate_json(frame)
+        return Message(
+            join_ref=join_ref, ref=ref, topic=topic, event=event, payload=payload
+        )
+    if not frame:
+        raise ValueError("empty binary frame")
+    if frame[0] != _USER_BROADCAST:
+        raise ValueError(f"unknown binary frame kind {frame[0]}")
+    return _decode_user_broadcast(frame)
+
+
+def _encode_user_broadcast_push(
+    message: Message, user_event: str, user_payload: bytes
+) -> bytes:
+    fields = {
+        "join_ref": (message.join_ref or "").encode(),
+        "ref": (message.ref or "").encode(),
+        "topic": message.topic.encode(),
+        "user_event": user_event.encode(),
+    }
+    for name, value in fields.items():
+        if len(value) > _MAX_FIELD_LENGTH:
+            raise ValueError(
+                f"{name} is {len(value)} bytes long, the maximum is {_MAX_FIELD_LENGTH}"
+            )
+    header = bytes(
+        [
+            _USER_BROADCAST_PUSH,
+            *(len(value) for value in fields.values()),
+            0,
+            _BINARY_ENCODING,
+        ]
+    )
+    return b"".join([header, *fields.values(), user_payload])
+
+
+def _decode_user_broadcast(frame: bytes) -> Message:
+    if len(frame) < _USER_BROADCAST_HEADER_LENGTH:
+        raise ValueError("truncated user broadcast frame")
+    topic_size, user_event_size, metadata_size, encoding = frame[1:5]
+    offset = _USER_BROADCAST_HEADER_LENGTH
+    topic, offset = _read(frame, offset, topic_size)
+    user_event, offset = _read(frame, offset, user_event_size)
+    metadata, offset = _read(frame, offset, metadata_size)
+    body = frame[offset:]
+
+    user_payload: JSON | bytes
+    if encoding == _JSON_ENCODING:
+        user_payload = json.loads(body)
+    elif encoding == _BINARY_ENCODING:
+        user_payload = body
+    else:
+        raise ValueError(f"unknown user broadcast payload encoding {encoding}")
+
+    payload: dict[str, JSON | bytes] = {
+        "type": ChannelEvents.broadcast.value,
+        "event": user_event.decode(),
+        "payload": user_payload,
+    }
+    if metadata:
+        payload["meta"] = json.loads(metadata)
+    return Message(
+        topic=topic.decode(), event=ChannelEvents.broadcast.value, payload=payload
+    )
+
+
+def _read(frame: bytes, offset: int, size: int) -> tuple[bytes, int]:
+    end = offset + size
+    if end > len(frame):
+        raise ValueError("truncated user broadcast frame")
+    return frame[offset:end], end

src/realtime/src/realtime/channel.py

@@ -3,7 +3,7 @@
 import asyncio
 import contextlib
 import logging
-from collections.abc import Sequence
+from collections.abc import Buffer, Sequence
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, Any, Self
 
@@ -305,7 +305,7 @@ def presence_state(self) -> RealtimePresenceState:
         return self.presence.state
 
     async def _send_event(
-        self, event: ChannelEvents, payload: Mapping[str, JSON]
+        self, event: ChannelEvents, payload: Mapping[str, JSON | bytes]
     ) -> ReplyMessage:
         message = Message(
             topic=self.topic,
@@ -317,7 +317,7 @@ async def _send_event(
         return await self.socket._send(message)
 
     async def _send_event_no_wait(
-        self, event: ChannelEvents, payload: Mapping[str, JSON]
+        self, event: ChannelEvents, payload: Mapping[str, JSON | bytes]
     ) -> None:
         message = Message(
             topic=self.topic,
@@ -329,19 +329,28 @@ async def _send_event_no_wait(
         return await self.socket._send_no_wait(message)
 
     # Broadcast methods
-    async def send_broadcast(self, event: str, data: JSON) -> ReplyMessage | None:
+    async def send_broadcast(
+        self, event: str, data: JSON | Buffer
+    ) -> ReplyMessage | None:
         """
         Send a broadcast message through this channel.
 
         Without ``broadcast_ack``, returning means the message was written to the
         connection, not that the server received it.
 
         :param event: The name of the broadcast event
-        :param data: The payload to broadcast
+        :param data: The payload to broadcast. A bytes-like object, such as
+            ``bytes``, ``bytearray`` or ``memoryview``, is sent as a binary frame,
+            and receivers get it as ``bytes`` in ``BroadcastPayload.payload``.
+            Anything else is sent as JSON.
+        :raises ValueError: If ``data`` is bytes-like and the topic or the event
+            is longer than 255 bytes in UTF-8.
         :raises websockets.ConnectionClosed: If the connection is closed.
         :raises Exception: If the connection already failed, the exception that
             failed it, such as a TimeoutError when a heartbeat timed out.
         """
+        if isinstance(data, Buffer):
+            data = bytes(data)
         payload = {"type": "broadcast", "event": event, "payload": data}
         if self.options.broadcast_ack:
             return await self._send_event(ChannelEvents.broadcast, payload)

src/realtime/src/realtime/client.py

@@ -5,12 +5,12 @@
 from warnings import warn
 
 import websockets
-from pydantic import ValidationError
 from supabase_utils.http.errors import SupabaseTransportError
 from websockets.asyncio.client import ClientConnection
 from yarl import URL
 
 from ._redact import RedactingLoggerAdapter
+from ._serializer import decode, encode
 from .channel import RealtimeChannel, RealtimeChannelOptions
 from .exceptions import MaxRetriesExceeded
 from .message import Message, ReplyMessage
@@ -27,6 +27,12 @@
 websockets_logger = RedactingLoggerAdapter(logging.getLogger("websockets.client"))
 
 
+def _loggable(frame: str | bytes) -> str:
+    if isinstance(frame, str):
+        return frame
+    return f"<binary frame of {len(frame)} bytes>"
+
+
 def normalize_url(url: URL, token: str | None) -> URL:
     if url.scheme not in {"ws", "wss", "http", "https"}:
         raise ValueError("url must be a valid WebSocket URL or HTTP URL string")
@@ -185,11 +191,13 @@ async def listen(self) -> None:
         """
         try:
             while True:
-                msg = await self._ws_connection.recv(decode=False)
+                frame = await self._ws_connection.recv()
                 try:
-                    message = Message.model_validate_json(msg)
-                except ValidationError as e:
-                    logger.error("Unrecognized message format %r\n%s", msg, e)
+                    message = decode(frame)
+                except ValueError as e:
+                    logger.error(
+                        "Unrecognized message format %s\n%s", _loggable(frame), e
+                    )
                     continue
                 if message.ref is not None and (
                     fut := self.message_refs.get(message.ref)
@@ -356,13 +364,13 @@ async def _send(self, message: Message) -> ReplyMessage:
         assert message.ref, "Cannot wait for reponse on a message without ref"
 
         self._raise_connection_failure()
-        message_str = message.model_dump_json()
-        logger.debug("sending: %s", message_str)
+        frame = encode(message)
+        logger.debug("sending: %s", _loggable(frame))
 
         receive: asyncio.Future[Message] = asyncio.Future()
         self.message_refs[message.ref] = receive
         try:
-            await self._ws_connection.send(message_str)
+            await self._ws_connection.send(frame)
             logger.debug("waiting for ref=%r", message.ref)
             response = await self._get_message_or_raise(receive)
         finally:
@@ -372,9 +380,9 @@ async def _send(self, message: Message) -> ReplyMessage:
 
     async def _send_no_wait(self, message: Message) -> None:
         self._raise_connection_failure()
-        message_str = message.model_dump_json()
-        logger.debug("sending: %s", message_str)
-        await self._ws_connection.send(message_str)
+        frame = encode(message)
+        logger.debug("sending: %s", _loggable(frame))
+        await self._ws_connection.send(frame)
 
     def _raise_connection_failure(self) -> None:
         failure = self._connection_failure

src/realtime/src/realtime/message.py

@@ -21,7 +21,7 @@ class Message(BaseModel):
     """
 
     event: str
-    payload: Mapping[str, JSON]
+    payload: Mapping[str, JSON | bytes]
     topic: str
     ref: str | None = None
     join_ref: str | None = None

src/realtime/src/realtime/types.py

@@ -5,12 +5,12 @@
 from enum import Enum
 from typing import Annotated, Any, Literal, TypeVar
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, Strict
 
 # Constants
 DEFAULT_TIMEOUT = 10
 PHOENIX_CHANNEL = "phoenix"
-PROTOCOL_VERSION = "1.0.0"
+PROTOCOL_VERSION = "2.0.0"
 DEFAULT_HEARTBEAT_INTERVAL = 25
 DEFAULT_HEARTBEAT_TIMEOUT = 5
 
@@ -131,7 +131,7 @@ class BroadcastMeta(BaseModel):
 
 class BroadcastPayload(BaseModel):
     event: str
-    payload: dict[str, Any]
+    payload: dict[str, Any] | Annotated[bytes, Strict()]
     type: str | None = None
     meta: BroadcastMeta | None = None
 

src/realtime/tests/test_connection.py

@@ -14,6 +14,7 @@
     RealtimeClient,
     RealtimePostgresChangesListenEvent,
 )
+from realtime._serializer import decode
 from realtime.channel import RealtimeChannelOptions
 from realtime.client import connect_once, exponential_delay, normalize_url
 from realtime.message import (
@@ -129,9 +130,24 @@ async def test_broadcast_events(socket: RealtimeClient) -> None:
         assert msg1.payload.event == "test-event"
         assert msg2.payload.event == "test-event"
         assert msg3.payload.event == "test-event"
-        assert msg1.payload.payload["message"] == "Event 1"
-        assert msg2.payload.payload["message"] == "Event 2"
-        assert msg3.payload.payload["message"] == "Event 3"
+        assert msg1.payload.payload == {"message": "Event 1"}
+        assert msg2.payload.payload == {"message": "Event 2"}
+        assert msg3.payload.payload == {"message": "Event 3"}
+
+
+@pytest.mark.asyncio
+async def test_binary_broadcast_events(socket: RealtimeClient) -> None:
+    channel_options = RealtimeChannelOptions().broadcast(listen_self=True, ack=True)
+    data = bytes(range(256))
+    async with socket.channel("test-binary-broadcast", channel_options) as chan:
+        reply = await chan.send_broadcast("binary-event", data)
+        assert reply is not None
+        assert reply.payload.status == "ok"
+
+        message = await chan.messages.__anext__()
+        assert isinstance(message, BroadcastMessage)
+        assert message.payload.event == "binary-event"
+        assert message.payload.payload == data
 
 
 @pytest.mark.asyncio
@@ -346,6 +362,7 @@ async def test_subscribe_to_private_channel_with_broadcast_replay(
         for i in range(messages):
             broadcast = await message_stream.__anext__()
             assert isinstance(broadcast, BroadcastMessage)
+            assert isinstance(broadcast.payload.payload, dict)
             assert broadcast.payload.payload["i"] == str(i)
             meta = broadcast.payload.meta
             assert meta
@@ -363,7 +380,7 @@ def test_normalize_url_includes_connection_params() -> None:
     assert dict(url.query) == {
         "apikey": "publishable-key",
         "log_level": "info",
-        "vsn": "1.0.0",
+        "vsn": "2.0.0",
     }
 
 
@@ -420,8 +437,8 @@ def fake_client() -> tuple[RealtimeClient, list[Message]]:
     ws = AsyncMock()
     client = RealtimeClient(URL("ws://localhost/realtime/v1"), ws)
 
-    async def send(raw: str) -> None:
-        message = Message.model_validate_json(raw)
+    async def send(frame: str | bytes) -> None:
+        message = decode(frame)
         sent.append(message)
         if message.ref is not None and message.event != ChannelEvents.access_token:
             reply = Message(