From eabf8973e5a912dd857528ca749c2309ac614e94 Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Mon, 28 Sep 2026 12:46:27 +0000 Subject: [PATCH] fix(realtime): retain configured queues without replaying attempted sends --- src/openai/_send_queue.py | 35 +++-- src/openai/resources/realtime/realtime.py | 20 +-- tests/lib/test_realtime_websocket_contract.py | 137 ++++++++++++++++++ tests/test_send_queue_reconnect.py | 69 +++++++-- 4 files changed, 222 insertions(+), 39 deletions(-) diff --git a/src/openai/_send_queue.py b/src/openai/_send_queue.py index 56d7824422..b3eb812b0e 100644 --- a/src/openai/_send_queue.py +++ b/src/openai/_send_queue.py @@ -37,11 +37,12 @@ def enqueue(self, data: str) -> None: self._queue.append((data, byte_length)) self._bytes += byte_length - def flush_sync(self, send: typing.Callable[[str], object]) -> None: + def flush_sync(self, send: typing.Callable[[str], object], *, requeue_failed: bool = True) -> None: """Send every queued message via *send*. If *send* raises, the failing message and all subsequent messages - are re-queued and the error is re-raised. + are re-queued and the error is re-raised. When `requeue_failed` is + false, release the attempted message even on failure or interruption. """ while isinstance(pending := self._begin_flush(), threading.Event): pending.wait() @@ -49,14 +50,21 @@ def flush_sync(self, send: typing.Callable[[str], object]) -> None: try: while pending: data, byte_length = pending[0] - send(data) - with self._lock: - pending.popleft() - self._bytes -= byte_length + sent = False + try: + send(data) + sent = True + finally: + if sent or not requeue_failed: + with self._lock: + pending.popleft() + self._bytes -= byte_length finally: self._end_flush(pending) - async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object]]) -> None: + async def flush_async( + self, send: typing.Callable[[str], typing.Awaitable[object]], *, requeue_failed: bool = True + ) -> None: """Async variant of :meth:`flush_sync`.""" while isinstance(pending := self._begin_flush(), threading.Event): # Waiting in a worker keeps the event loop responsive. Cancellation @@ -66,10 +74,15 @@ async def flush_async(self, send: typing.Callable[[str], typing.Awaitable[object try: while pending: data, byte_length = pending[0] - await send(data) - with self._lock: - pending.popleft() - self._bytes -= byte_length + sent = False + try: + await send(data) + sent = True + finally: + if sent or not requeue_failed: + with self._lock: + pending.popleft() + self._bytes -= byte_length finally: self._end_flush(pending) diff --git a/src/openai/resources/realtime/realtime.py b/src/openai/resources/realtime/realtime.py index aef7757cb0..f541bdd886 100644 --- a/src/openai/resources/realtime/realtime.py +++ b/src/openai/resources/realtime/realtime.py @@ -294,7 +294,7 @@ def __init__( self._extra_headers = extra_headers self._intentionally_closed = False self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=False) self.session = AsyncRealtimeSessionResource(self) @@ -373,11 +373,7 @@ async def send(self, event: RealtimeClientEvent | RealtimeClientEventParam) -> N if self._is_reconnecting: self._send_queue.enqueue(data) return - try: - await self._connection.send(data) - except Exception: - self._send_queue.enqueue(data) - raise + await self._connection.send(data) async def send_raw(self, data: bytes | str) -> None: if self._is_reconnecting: @@ -484,7 +480,7 @@ async def _send(data: str) -> None: await self._connection.send(data) try: - await self._send_queue.flush_async(_send) + await self._send_queue.flush_async(_send, requeue_failed=False) except Exception: log.warning("Failed to flush send queue after reconnect") @@ -818,7 +814,7 @@ def __init__( self._extra_headers = extra_headers self._intentionally_closed = False self._is_reconnecting = False - self._send_queue = send_queue or SendQueue() + self._send_queue = send_queue if send_queue is not None else SendQueue() self._event_handler_registry = EventHandlerRegistry(use_lock=True) self.session = RealtimeSessionResource(self) @@ -897,11 +893,7 @@ def send(self, event: RealtimeClientEvent | RealtimeClientEventParam) -> None: if self._is_reconnecting: self._send_queue.enqueue(data) return - try: - self._connection.send(data) - except Exception: - self._send_queue.enqueue(data) - raise + self._connection.send(data) def send_raw(self, data: bytes | str) -> None: if self._is_reconnecting: @@ -1002,7 +994,7 @@ def _reconnect(self, exc: Exception) -> bool: def _flush_send_queue(self) -> None: """Send all queued messages over the current connection.""" try: - self._send_queue.flush_sync(lambda data: self._connection.send(data)) + self._send_queue.flush_sync(lambda data: self._connection.send(data), requeue_failed=False) except Exception: log.warning("Failed to flush send queue after reconnect") diff --git a/tests/lib/test_realtime_websocket_contract.py b/tests/lib/test_realtime_websocket_contract.py index 5dddcd2707..d1dde5f00c 100644 --- a/tests/lib/test_realtime_websocket_contract.py +++ b/tests/lib/test_realtime_websocket_contract.py @@ -6,9 +6,13 @@ import httpx2 import pytest +from websockets.sync.client import ClientConnection from websockets.sync.server import ServerConnection +from websockets.asyncio.client import ClientConnection as AsyncClientConnection from openai import OpenAI, AsyncOpenAI +from openai._exceptions import WebSocketQueueFullError +from openai.types.websocket_reconnection import ReconnectingEvent from openai.types.realtime.realtime_error_event import RealtimeErrorEvent from openai.types.realtime.input_audio_buffer_cleared_event import InputAudioBufferClearedEvent @@ -133,3 +137,136 @@ def script(socket: ServerConnection) -> None: "event_id": "future-1", "text": "東京🙂", } + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_realtime_empty_manager_queue_keeps_budget_after_open(mode: str) -> None: + opened = 0 + + def script(socket: ServerConnection) -> None: + nonlocal opened + opened += 1 + if opened == 1: + socket.close(code=1011, reason="synthetic restart") + return + # Let the iterator resume even if its connection lost the manager's + # queue. Server-side inspection still proves exactly what was sent. + socket.send('{"type": "input_audio_buffer.cleared", "event_id": "ready"}') + assert json.loads(socket.recv(timeout=5)) == {"type": "input_audio_buffer.clear", "event_id": "queued"} + + with script_server(script, expected_connections=2) as url: + if mode == "sync": + with OpenAI( + api_key="fake-realtime-key", base_url=url, http_client=httpx2.Client(trust_env=False) + ) as client: + + def on_retry(_event: ReconnectingEvent) -> None: + manager.send({"type": "input_audio_buffer.clear", "event_id": "queued"}) + with pytest.raises(WebSocketQueueFullError): + manager.send({"type": "input_audio_buffer.append", "audio": "AAAA" * 100}) + + manager = client.realtime.connect( + model="gpt-realtime", max_queue_size=96, on_reconnecting=on_retry, initial_delay=0 + ) + with manager as connection: + event = next(iter(connection)) + assert event.type == "input_audio_buffer.cleared" + else: + async with AsyncOpenAI( + api_key="fake-realtime-key", base_url=url, http_client=httpx2.AsyncClient(trust_env=False) + ) as client: + + def on_async_retry(_event: ReconnectingEvent) -> None: + async_manager.send({"type": "input_audio_buffer.clear", "event_id": "queued"}) + with pytest.raises(WebSocketQueueFullError): + async_manager.send({"type": "input_audio_buffer.append", "audio": "AAAA" * 100}) + + async_manager = client.realtime.connect( + model="gpt-realtime", max_queue_size=96, on_reconnecting=on_async_retry, initial_delay=0 + ) + async with async_manager as connection: + event = await asyncio.wait_for(anext(aiter(connection)), 5) + assert event.type == "input_audio_buffer.cleared" + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("preopen", [True, False], ids=["preopen-flush", "direct"]) +async def test_realtime_recovery_never_replays_an_attempted_command( + mode: str, preopen: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + opened = 0 + recorded: list[tuple[int, str]] = [] + + def script(socket: ServerConnection) -> None: + nonlocal opened + opened += 1 + current = opened + if current == 1: + for _ in range(2 if preopen else 1): + recorded.append((current, json.loads(socket.recv(timeout=5))["event_id"])) + socket.close(code=1011, reason="synthetic restart") + return + socket.send('{"type": "input_audio_buffer.cleared", "event_id": "ready"}') + for _ in range(2 if preopen else 1): + recorded.append((current, json.loads(socket.recv(timeout=5))["event_id"])) + + # Fail after the real wire send: an exception cannot establish non-delivery. + original = ClientConnection.send + original_async = AsyncClientConnection.send + + def send_then_interrupt(self: ClientConnection, message: object, *args: object, **kwargs: object) -> None: + original(self, message, *args, **kwargs) # type: ignore[arg-type] + if json.loads(message)["event_id"] == "attempted": # type: ignore[arg-type] + raise OSError("synthetic interruption after send") + + async def async_send_then_interrupt( + self: AsyncClientConnection, message: object, *args: object, **kwargs: object + ) -> None: + await original_async(self, message, *args, **kwargs) # type: ignore[arg-type] + if json.loads(message)["event_id"] == "attempted": # type: ignore[arg-type] + raise OSError("synthetic interruption after send") + + if mode == "sync": + monkeypatch.setattr(ClientConnection, "send", send_then_interrupt) + else: + monkeypatch.setattr(AsyncClientConnection, "send", async_send_then_interrupt) + with script_server(script, expected_connections=2) as url: + if mode == "sync": + with OpenAI( + api_key="fake-realtime-key", base_url=url, http_client=httpx2.Client(trust_env=False) + ) as client: + + def on_retry(_event: ReconnectingEvent) -> None: + manager.send({"type": "input_audio_buffer.clear", "event_id": "during-recovery"}) + + manager = client.realtime.connect(model="gpt-realtime", on_reconnecting=on_retry, initial_delay=0) + if preopen: + for identity in ("first", "attempted", "unattempted"): + manager.send({"type": "input_audio_buffer.clear", "event_id": identity}) + with manager as connection: + if not preopen: + with pytest.raises(OSError, match="synthetic interruption after send"): + connection.send({"type": "input_audio_buffer.clear", "event_id": "attempted"}) + assert next(iter(connection)).type == "input_audio_buffer.cleared" + else: + async with AsyncOpenAI( + api_key="fake-realtime-key", base_url=url, http_client=httpx2.AsyncClient(trust_env=False) + ) as client: + + def on_async_retry(_event: ReconnectingEvent) -> None: + async_manager.send({"type": "input_audio_buffer.clear", "event_id": "during-recovery"}) + + async_manager = client.realtime.connect( + model="gpt-realtime", on_reconnecting=on_async_retry, initial_delay=0 + ) + if preopen: + for identity in ("first", "attempted", "unattempted"): + async_manager.send({"type": "input_audio_buffer.clear", "event_id": identity}) + async with async_manager as connection: + if not preopen: + with pytest.raises(OSError, match="synthetic interruption after send"): + await connection.send({"type": "input_audio_buffer.clear", "event_id": "attempted"}) + event = await asyncio.wait_for(anext(aiter(connection)), 5) + assert event.type == "input_audio_buffer.cleared" + expected = [(1, "first"), (1, "attempted"), (2, "unattempted"), (2, "during-recovery")] + assert recorded == (expected if preopen else [(1, "attempted"), (2, "during-recovery")]) diff --git a/tests/test_send_queue_reconnect.py b/tests/test_send_queue_reconnect.py index fec12479ab..5d48ed5097 100644 --- a/tests/test_send_queue_reconnect.py +++ b/tests/test_send_queue_reconnect.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from unittest.mock import AsyncMock, MagicMock import pytest @@ -22,15 +23,17 @@ def test_reconnect_retries_bounded_send_queue( q.enqueue("aaa") ws = MagicMock() attempts = 0 + realtime = connection_type is RealtimeConnection def failing_send(data: str) -> None: nonlocal attempts - assert data == "aaa" + assert data == ("b" if realtime and attempts == 1 else "aaa") if attempts == 0: q.enqueue("b") attempts += 1 - with pytest.raises(WebSocketQueueFullError): - q.enqueue("c") + if not realtime or attempts == 1: + with pytest.raises(WebSocketQueueFullError): + q.enqueue("c") raise RuntimeError("fake send failure") ws.send.side_effect = failing_send @@ -44,10 +47,10 @@ def failing_send(data: str) -> None: ) for expected_attempts in range(1, 4): assert connection._reconnect(RuntimeError("fake disconnect")) - assert q._bytes == 4 - assert attempts == expected_attempts + assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4) + assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts) assert not connection._reconnect(RuntimeError("fake disconnect")) - assert attempts == 3 + assert attempts == (2 if realtime else 3) # A healthy application event resets the budget; an upgrade alone does not. ws.recv.return_value = '{"type": "response.created"}' @@ -56,7 +59,7 @@ def failing_send(data: str) -> None: sent: list[str] = [] ws.send.side_effect = sent.append assert connection._reconnect(RuntimeError("fake disconnect")) - assert sent == ["aaa", "b"] + assert sent == ([] if realtime else ["aaa", "b"]) assert q._bytes == 0 for _ in range(2): assert connection._reconnect(RuntimeError("fake disconnect")) @@ -76,15 +79,17 @@ async def test_async_reconnect_retries_bounded_send_queue( q.enqueue("aaa") ws = MagicMock() attempts = 0 + realtime = connection_type is AsyncRealtimeConnection async def failing_send(data: str) -> None: nonlocal attempts - assert data == "aaa" + assert data == ("b" if realtime and attempts == 1 else "aaa") if attempts == 0: q.enqueue("b") attempts += 1 - with pytest.raises(WebSocketQueueFullError): - q.enqueue("c") + if not realtime or attempts == 1: + with pytest.raises(WebSocketQueueFullError): + q.enqueue("c") raise RuntimeError("fake send failure") ws.send = AsyncMock(side_effect=failing_send) @@ -98,10 +103,10 @@ async def failing_send(data: str) -> None: ) for expected_attempts in range(1, 4): assert await connection._reconnect(RuntimeError("fake disconnect")) - assert q._bytes == 4 - assert attempts == expected_attempts + assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4) + assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts) assert not await connection._reconnect(RuntimeError("fake disconnect")) - assert attempts == 3 + assert attempts == (2 if realtime else 3) # A healthy application event resets the budget; an upgrade alone does not. ws.recv = AsyncMock(return_value='{"type": "response.created"}') @@ -110,8 +115,44 @@ async def failing_send(data: str) -> None: sent: list[str] = [] ws.send.side_effect = sent.append assert await connection._reconnect(RuntimeError("fake disconnect")) - assert sent == ["aaa", "b"] + assert sent == ([] if realtime else ["aaa", "b"]) assert q._bytes == 0 for _ in range(2): assert await connection._reconnect(RuntimeError("fake disconnect")) assert not await connection._reconnect(RuntimeError("retry budget exhausted")) + + +@pytest.mark.asyncio +async def test_cancelled_realtime_flush_does_not_replay_active_send() -> None: + q = SendQueue(max_bytes=4) + q.enqueue("é") + q.enqueue("b") + active = asyncio.Event() + socket = MagicMock() + + async def suspended_send(data: str) -> None: + assert data == "é" + active.set() + await asyncio.Event().wait() + + socket.send = AsyncMock(side_effect=suspended_send) + connection = AsyncRealtimeConnection(socket, send_queue=q) + task = asyncio.create_task(connection._flush_send_queue()) + try: + await asyncio.wait_for(active.wait(), 5) + q.enqueue("c") + with pytest.raises(WebSocketQueueFullError): + q.enqueue("d") + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # Only the active send was attempted; its bytes are released, and the + # queued tail and concurrent enqueue remain in order for the next socket. + q.enqueue("dd") + sent: list[str] = [] + socket.send.side_effect = sent.append + await connection._flush_send_queue() + assert sent == ["b", "c", "dd"] + q.enqueue("1234")