From a444e4256d9b9a2349d187f792dc4bfa2dfba533 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Mon, 28 Sep 2026 04:56:06 -0700 Subject: [PATCH 1/2] test: verify Live grouper disposal on a shared Python WebSocket (#3977) When two transcript groupers observe the same Live dispatcher, callers must be able to detach and close one without stopping the other or losing the event stream. The existing grouper tests covered its own listeners and timers but did not exercise an attached connection. This tests both sync and async clients over a real local WebSocket. It checks that the first grouper cancels its active timer, typed and wildcard observers keep receiving the original values and future metadata, the second grouper continues accumulating, and the same socket accepts a post-disposal update. The peer waits for explicit caller teardown. Tests only; no generated or API changes. Validation: - 238 Live tests passed on Pydantic 2.12.5 and 238 on 1.10.26. - After typing the close callbacks and checking only event IDs present on the wire: 14 socket-role and grouper tests pass; Pyright reports 0 errors on the affected file. - Repository-configured `mypy .` passes across 1,920 source files; Ruff check, format and diff checks pass. --- tests/lib/live/test_websocket_roles.py | 143 ++++++++++++++++++++++++- 1 file changed, 142 insertions(+), 1 deletion(-) diff --git a/tests/lib/live/test_websocket_roles.py b/tests/lib/live/test_websocket_roles.py index e3b0ae5cdc..40de79fed2 100644 --- a/tests/lib/live/test_websocket_roles.py +++ b/tests/lib/live/test_websocket_roles.py @@ -9,10 +9,17 @@ from websockets.sync.server import ServerConnection from openai import OpenAI, AsyncOpenAI -from openai.types.live import SessionStartedEvent, SessionUpdatedEvent, OutputTranscriptDeltaEvent +from openai.types.live import ( + ServerEvent, + SessionClosedEvent, + SessionStartedEvent, + SessionUpdatedEvent, + OutputTranscriptDeltaEvent, +) from openai.resources.live.live import LiveConnection, AsyncLiveConnection from openai.resources.live.forks import ForksConnection, AsyncForksConnection +from .helpers import FakeClock, Recording, ClockGrouper, AsyncClockGrouper from ..responses.test_websocket_session import script_server @@ -157,3 +164,137 @@ def script(socket: ServerConnection) -> None: assert isinstance(async_updated, SessionUpdatedEvent) assert async_updated.client_event_id == "caller-update" assert async_updated.session.id == "live_fixture" + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_disposed_grouper_leaves_dispatcher_other_observers_and_socket_usable(mode: str) -> None: + session = {"id": "live_fixture", "model": "gpt-live-1", "status": "active", "expires_at": 123} + first_clock, second_clock = FakeClock(), FakeClock() + transcripts = [ + { + "type": "session.output_transcript.delta", + "event_id": f"part-{index}", + "delta": value, + "start_ms": index * 200, + "end_ms": (index + 1) * 200, + } + for index, value in enumerate(["One", " two", " three", " four"]) + ] + observed: list[ServerEvent] = [] + typed: list[OutputTranscriptDeltaEvent] = [] + + def script(socket: ServerConnection) -> None: + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.start", + "event_id": "caller-start", + "session": {"model": "gpt-live-1"}, + } + socket.send(json.dumps({"type": "session.started", "event_id": "started", "session": session})) + for event in transcripts[:2]: + socket.send(json.dumps(event)) + # Issued only after the caller detaches and closes its first grouper. + assert json.loads(socket.recv(timeout=5)) == { + "type": "session.update", + "event_id": "after-dispose", + "session": {}, + } + socket.send( + json.dumps( + { + "type": "session.updated", + "event_id": "updated", + "client_event_id": "after-dispose", + "session": session, + "future_metadata": {"explicit_null": None, "nested": [1, "retained"]}, + } + ) + ) + for event in transcripts[2:]: + socket.send(json.dumps(event)) + assert json.loads(socket.recv(timeout=5)) == {"type": "session.close", "event_id": "caller-finish"} + socket.send(json.dumps({"type": "session.closed", "event_id": "closed", "reason": "client_close"})) + # The shared fixture rejects extra writes/reconnects and waits for caller close. + + with script_server(script) as url: + if mode == "sync": + first = ClockGrouper(first_clock) + second = ClockGrouper(second_clock) + first_record, second_record = Recording(first, first_clock), Recording(second, second_clock) + with OpenAI(api_key="ek_fake_live", base_url=url, http_client=httpx2.Client(trust_env=False)) as client: + with client.live.connect() as connection: + connection.on("session.output_transcript.delta", first.push) + connection.on("session.output_transcript.delta", second.push) + connection.on("session.output_transcript.delta", typed.append) + connection.on("event", observed.append) + + def manage(event: OutputTranscriptDeltaEvent) -> None: + if event.event_id == "part-1": + assert first_clock.pending + connection.off("session.output_transcript.delta", first.push) + first.close() + first.close() + assert not first_clock.pending + connection.session.update(session={}, event_id="after-dispose") + elif event.event_id == "part-3": + connection.session.close(event_id="caller-finish") + + def close_connection(_event: SessionClosedEvent) -> None: + connection.close() + + connection.on("session.output_transcript.delta", manage) + connection.on("session.closed", close_connection) + connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + connection.dispatch_events() + second.close() + else: + async_first = AsyncClockGrouper(first_clock) + async_second = AsyncClockGrouper(second_clock) + first_record, second_record = ( + Recording(async_first, first_clock), + Recording(async_second, second_clock), + ) + async with AsyncOpenAI( + api_key="ek_fake_live", base_url=url, http_client=httpx2.AsyncClient(trust_env=False) + ) as async_client: + async with async_client.live.connect() as async_connection: + async_connection.on("session.output_transcript.delta", async_first.push) + async_connection.on("session.output_transcript.delta", async_second.push) + async_connection.on("session.output_transcript.delta", typed.append) + async_connection.on("event", observed.append) + + async def async_manage(event: OutputTranscriptDeltaEvent) -> None: + if event.event_id == "part-1": + assert first_clock.pending + async_connection.off("session.output_transcript.delta", async_first.push) + await async_first.close() + await async_first.close() + assert not first_clock.pending + await async_connection.session.update(session={}, event_id="after-dispose") + elif event.event_id == "part-3": + await async_connection.session.close(event_id="caller-finish") + + async def close_async_connection(_event: SessionClosedEvent) -> None: + await async_connection.close() + + async_connection.on("session.output_transcript.delta", async_manage) + async_connection.on("session.closed", close_async_connection) + await async_connection.session.start(session={"model": "gpt-live-1"}, event_id="caller-start") + await asyncio.wait_for(async_connection.dispatch_events(), timeout=5) + await async_second.close() + + assert [event.to_dict().get("event_id") for event in observed] == [ + "started", + "part-0", + "part-1", + "updated", + "part-2", + "part-3", + "closed", + ] + assert [event.to_dict() for event in typed] == transcripts + updated = observed[3] + assert isinstance(updated, SessionUpdatedEvent) + assert updated.to_dict()["future_metadata"] == {"explicit_null": None, "nested": [1, "retained"]} + assert [(event.segment.text, event.reason) for event in first_record.closed] == [("One two", "manual")] + assert [(event.segment.text, event.reason) for event in second_record.closed] == [("One two three four", "manual")] + assert not first_clock.pending and not second_clock.pending From a52805c2537422ad602eba2bf66072880ea3fd22 Mon Sep 17 00:00:00 2001 From: Mark Stuart Date: Mon, 28 Sep 2026 06:25:42 -0700 Subject: [PATCH 2/2] fix(realtime): retain configured queues without replaying attempted sends (#3978) An empty Realtime manager queue was replaced on connect, losing its configured byte limit and later messages. A typed or queued send could also be replayed on reconnect even after reaching the server. Preserve the manager's queue and release only attempted writes on failure or cancellation; never-attempted messages remain in order. Other endpoints retain the shared queue's existing defaults. Six sync/async real-socket regressions fail on the base and pass with this change, covering empty queues and interrupted pre-open/direct sends. A focused cancellation regression verifies in-flight bounds and retained messages. Validation: 456 affected tests per Pydantic 1/2, 45 final queue/Realtime tests per version, Ruff, mypy (1,865 files), Pyright, and the trusted custom-code check (7,766/10,000). --- 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")