None of these were async either

Author: blast-hardcheese

src/replit_river/client_session.py

@@ -140,7 +140,7 @@ async def _handle_messages_from_ws(self) -> None:
                         case other:
                             assert_never(other)
 
-                    await self._buffer.remove_old_messages(
+                    self._buffer.remove_old_messages(
                         self._seq_manager.receiver_ack,
                     )
                     self._reset_session_close_countdown()

src/replit_river/message_buffer.py

@@ -20,10 +20,6 @@ def __init__(self, max_num_messages: int = MAX_MESSAGE_BUFFER_SIZE):
         self._space_available_cond = asyncio.Condition()
         self._closed = False
 
-    async def empty(self) -> bool:
-        """Check if the buffer is empty"""
-        return len(self.buffer) == 0
-
     async def put(self, message: TransportMessage) -> None:
         """Add a message to the buffer. Blocks until there is space in the buffer.
 
@@ -38,19 +34,19 @@ async def put(self, message: TransportMessage) -> None:
                 raise MessageBufferClosedError("message buffer is closed")
             self.buffer.append(message)
 
-    async def peek(self) -> TransportMessage | None:
+    def peek(self) -> TransportMessage | None:
         """Peek the first message in the buffer, returns None if the buffer is empty."""
         if len(self.buffer) == 0:
             return None
         return self.buffer[0]
 
-    async def remove_old_messages(self, min_seq: int) -> None:
+    def remove_old_messages(self, min_seq: int) -> None:
         """Remove messages in the buffer with a seq number less than min_seq."""
         self.buffer = [msg for msg in self.buffer if msg.seq >= min_seq]
         async with self._space_available_cond:
             self._space_available_cond.notify_all()
 
-    async def close(self) -> None:
+    def close(self) -> None:
         """
         Closes the message buffer and rejects any pending put operations.
         """

src/replit_river/server_session.py

@@ -136,7 +136,7 @@ async def _handle_messages_from_ws(self, tg: asyncio.TaskGroup) -> None:
                             pass
                         case other:
                             assert_never(other)
-                    await self._buffer.remove_old_messages(
+                    self._buffer.remove_old_messages(
                         self._seq_manager.receiver_ack,
                     )
                     self._reset_session_close_countdown()

src/replit_river/session.py

@@ -229,7 +229,7 @@ async def get_next_expected_seq(self) -> int:
 
     async def get_next_sent_seq(self) -> int:
         """Get the next sequence number that the client will send."""
-        nextMessage = await self._buffer.peek()
+        nextMessage = self._buffer.peek()
         if nextMessage:
             return nextMessage.seq
         return self._seq_manager.get_seq()
@@ -325,7 +325,7 @@ async def close(self) -> None:
 
             await self.close_websocket(self._ws_wrapper, should_retry=False)
 
-            await self._buffer.close()
+            self._buffer.close()
 
             # Clear the session in transports
             await self._close_session_callback(self)

tests/test_message_buffer.py

@@ -44,7 +44,7 @@ async def put_messages() -> None:
         # Wait for the put call to return.
         await sync_events.get()
         assert len(buffer.buffer) == 1
-        await buffer.remove_old_messages(i)
+        buffer.remove_old_messages(i)
 
     await background_puts
 
@@ -57,7 +57,7 @@ async def test_message_buffer_close() -> None:
     buffer = MessageBuffer(max_num_messages=1)
     await buffer.put(mock_transport_message(seq=1))
     background_put = asyncio.create_task(buffer.put(mock_transport_message(seq=1)))
-    await buffer.close()
+    buffer.close()
     with pytest.raises(MessageBufferClosedError):
         await background_put
     with pytest.raises(MessageBufferClosedError):

tests/test_seq_manager.py

@@ -1,9 +1,7 @@
-import asyncio
-
 import pytest
 
 from replit_river.seq_manager import (
-    IgnoreMessageException,
+    IgnoreMessage,
     OutOfOrderMessageException,
     SeqManager,
 )
@@ -14,17 +12,17 @@
 @pytest.mark.asyncio
 async def test_initial_sequence_and_ack_numbers(no_logging_error: NoErrors) -> None:
     manager = SeqManager()
-    assert await manager.get_seq() == 0, "Initial sequence number should be 0"
-    assert await manager.get_ack() == 0, "Initial acknowledgment number should be 0"
+    assert manager.get_seq() == 0, "Initial sequence number should be 0"
+    assert manager.get_ack() == 0, "Initial acknowledgment number should be 0"
     no_logging_error()
 
 
 @pytest.mark.asyncio
 async def test_sequence_number_increment(no_logging_error: NoErrors) -> None:
     manager = SeqManager()
-    initial_seq = await manager.get_seq_and_increment()
+    initial_seq = manager.get_seq_and_increment()
     assert initial_seq == 0, "Sequence number should start at 0"
-    new_seq = await manager.get_seq()
+    new_seq = manager.get_seq()
     assert new_seq == 1, "Sequence number should increment to 1"
     no_logging_error()
 
@@ -33,42 +31,40 @@ async def test_sequence_number_increment(no_logging_error: NoErrors) -> None:
 async def test_message_reception(no_logging_error: NoErrors) -> None:
     manager = SeqManager()
     msg = transport_message(seq=0, ack=0, from_="client")
-    await manager.check_seq_and_update(
+    manager.check_seq_and_update(
         msg
     )  # No error should be raised for the correct sequence
-    assert await manager.get_ack() == 1, "Acknowledgment should be set to 1"
+    assert manager.get_ack() == 1, "Acknowledgment should be set to 1"
 
     # We assert no errors before we send out-of-order messages
     no_logging_error()
 
     # Test duplicate message
-    with pytest.raises(IgnoreMessageException):
-        await manager.check_seq_and_update(msg)
+    assert isinstance(manager.check_seq_and_update(msg), IgnoreMessage)
 
     # Test out of order message
     msg.seq = 2
     with pytest.raises(OutOfOrderMessageException):
-        await manager.check_seq_and_update(msg)
+        manager.check_seq_and_update(msg)
 
 
 @pytest.mark.asyncio
 async def test_acknowledgment_setting(no_logging_error: NoErrors) -> None:
     manager = SeqManager()
     msg = transport_message(seq=0, ack=0, from_="client")
-    await manager.check_seq_and_update(msg)
-    assert await manager.get_ack() == 1, "Acknowledgment number should be updated"
+    manager.check_seq_and_update(msg)
+    assert manager.get_ack() == 1, "Acknowledgment number should be updated"
     no_logging_error()
 
 
 @pytest.mark.asyncio
 async def test_concurrent_access_to_sequence(no_logging_error: NoErrors) -> None:
     manager = SeqManager()
-    tasks = [manager.get_seq_and_increment() for _ in range(10)]
-    results = await asyncio.gather(*tasks)
+    results = [manager.get_seq_and_increment() for _ in range(10)]
     assert len(set(results)) == 10, (
         "Each increment call should return a unique sequence number"
     )
-    assert await manager.get_seq() == 10, (
+    assert manager.get_seq() == 10, (
         "Final sequence number should be 10 after 10 increments"
     )
     no_logging_error()