Remove prefix bytes shenanigans (#88)

Why
===

There was a world where we muxed crosis/pid1 and river/pid2 over one
websocket, so we had some prefix bytes. We killed that months ago. Just
cleaning up.

What changed
============

:yeet:

Test plan
=========

river rivers

Author: masad-frost

replit_river/client_transport.py

@@ -250,7 +250,6 @@ async def websocket_closed_callback() -> None:
                     payload=handshake_request.model_dump(),
                 ),
                 ws=websocket,
-                prefix_bytes=self._transport_options.get_prefix_bytes(),
                 websocket_closed_callback=websocket_closed_callback,
             )
             return handshake_request

replit_river/messages.py

@@ -31,23 +31,19 @@ class FailedSendingMessageException(Exception):
 
 PROTOCOL_VERSION = "v1.1"
 
-CROSIS_PREFIX_BYTES = b"\x00\x00"
-PID2_PREFIX_BYTES = b"\xff\xff"
-
 
 async def send_transport_message(
     msg: TransportMessage,
     ws: WebSocketCommonProtocol,
     websocket_closed_callback: Callable[[], Coroutine[Any, Any, None]],
-    prefix_bytes: bytes = b"",
 ) -> None:
     logger.debug("sending a message %r to ws %s", msg, ws)
     try:
         packed = msgpack.packb(
             msg.model_dump(by_alias=True, exclude_none=True), datetime=True
         )
         assert isinstance(packed, bytes)
-        await ws.send(prefix_bytes + packed)
+        await ws.send(packed)
     except websockets.exceptions.ConnectionClosed as e:
         await websocket_closed_callback()
         raise WebsocketClosedException("Websocket closed during send message") from e
@@ -73,15 +69,6 @@ def parse_transport_msg(
         raise IgnoreMessageException(
             f"ignored a message beacuse it was a text frame: {message}"
         )
-    if transport_options.use_prefix_bytes:
-        if message.startswith(CROSIS_PREFIX_BYTES):
-            raise IgnoreMessageException("Skip crosis message")
-        elif message.startswith(PID2_PREFIX_BYTES):
-            message = message[len(PID2_PREFIX_BYTES) :]
-        else:
-            raise InvalidMessageException(
-                f"Got message without prefix bytes: {formatted_bytes(message)[:5]}"
-            )
     try:
         # :param int timestamp:
         #     Control how timestamp type is unpacked:

replit_river/session.py

@@ -317,7 +317,6 @@ async def _send_buffered_messages(
                 await self._send_transport_message(
                     msg,
                     websocket,
-                    prefix_bytes=self._transport_options.get_prefix_bytes(),
                 )
             except WebsocketClosedException:
                 logger.info(
@@ -332,11 +331,10 @@ async def _send_transport_message(
         self,
         msg: TransportMessage,
         websocket: websockets.WebSocketCommonProtocol,
-        prefix_bytes: bytes = b"",
     ) -> None:
         try:
             await send_transport_message(
-                msg, websocket, self._begin_close_session_countdown, prefix_bytes
+                msg, websocket, self._begin_close_session_countdown
             )
         except WebsocketClosedException as e:
             raise e
@@ -401,7 +399,6 @@ async def send_message(
                 await self._send_transport_message(
                     msg,
                     self._ws_wrapper.ws,
-                    prefix_bytes=self._transport_options.get_prefix_bytes(),
                 )
         except WebsocketClosedException as e:
             logger.debug(

replit_river/transport_options.py

@@ -3,8 +3,6 @@
 
 from pydantic import BaseModel
 
-CROSIS_PREFIX_BYTES = b"\x00\x00"
-PID2_PREFIX_BYTES = b"\xff\xff"
 MAX_MESSAGE_BUFFER_SIZE = 1024
 
 
@@ -25,15 +23,11 @@ class TransportOptions(BaseModel):
     heartbeat_ms: float = 2_500
     # TODO: This should have a better name like max_failed_heartbeats
     heartbeats_until_dead: int = 4
-    use_prefix_bytes: bool = False
     close_session_check_interval_ms: float = 100
     connection_retry_options: ConnectionRetryOptions = ConnectionRetryOptions()
     buffer_size: int = 1_000
     transparent_reconnect: bool = True
 
-    def get_prefix_bytes(self) -> bytes:
-        return PID2_PREFIX_BYTES if self.use_prefix_bytes else b""
-
     def websocket_disconnect_grace_ms(self) -> float:
         return self.heartbeat_ms * self.heartbeats_until_dead