From 1c97767093247fa5a80338d4d6c7c01182d64dd3 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Fri, 17 Apr 2026 15:16:10 -0700 Subject: [PATCH 1/2] fix(phx_channel): buffer pushes that arrive before on() is registered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first CI run on main failed flakily (2/641 tests on Python 3.11, 1/641 on 3.12) on "on__delivers_contract_valid_payloads" tests: await client.register_scenario({ "topic": ..., "onJoin": [{"type": "autoReply"}, {"type": "autoPush", "event": E}], }) channel = await Channel.join_x(socket, ...) # (1) channel.on(E, handler) # (2) await asyncio.wait_for(future, timeout=1.0) The server's onJoin scenario fires the join reply AND the autoPush back-to-back. The client's join future resolves at (1), then (2) runs, but on slower runners the autoPush frame can already be on the asyncio queue and dispatched before (2) registers the handler — so the push hits Channel._dispatch_event with an empty handler list and gets silently dropped. Local macOS was fast enough to dodge this; GHA runners aren't. Fix: if a user-event push arrives with no handlers yet, buffer it in `_pending_pushes[event]`. When `on(event, cb)` is later called, replay any buffered payloads into the callback in arrival order. Bounded at 32 events/type to keep orphaned topics from leaking unbounded memory. Matches Phoenix-client behavior in other languages where the handler is always registered before `join()` — our generated Python API conflates create+join into a single `.join_x()` class method, so without buffering, post-join handler registration inherits the race. Added two unit tests covering replay order + bounded buffer. Full contract suite (641 tests) passes locally across 10 consecutive runs. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/phx_channel/channel.py | 46 ++++++++++++++++++++++++------ src/phx_channel/tests/test_unit.py | 29 +++++++++++++++++++ 2 files changed, 67 insertions(+), 8 deletions(-) diff --git a/src/phx_channel/channel.py b/src/phx_channel/channel.py index 8fa5bac..e4747ae 100644 --- a/src/phx_channel/channel.py +++ b/src/phx_channel/channel.py @@ -37,6 +37,14 @@ def __init__(self, socket: Socket, topic: str, params: dict[str, Any]): self._event_handlers: dict[str, list[Callable[..., Any]]] = {} self._pending_replies: dict[str, asyncio.Future[dict[str, Any]]] = {} self._push_buffer: list[tuple[str, Any, asyncio.Future[dict[str, Any]]]] = [] + # Inbound pushes that arrived before any handler was registered. + # Server scenarios can `autoPush` in response to a join frame, and + # that push can land on the asyncio queue before the caller has + # had a chance to register a handler post-join. We replay these + # to the first matching handler registered within the window. + # Bounded so a forgotten handler doesn't leak unbounded memory. + self._pending_pushes: dict[str, list[Any]] = {} + self._pending_pushes_cap: int = 32 @property def topic(self) -> str: @@ -195,11 +203,24 @@ def on(self, event: str, callback: Callable[..., Any]) -> Callable[[], None]: """ Register a callback for a channel event. - Returns an unsubscribe function. + Returns an unsubscribe function. If pushes for this event already + arrived before any handler was registered (e.g. server-side + autoPush fired on join), they're replayed to this callback in + arrival order so the caller can't miss them due to scheduling. """ handlers = self._event_handlers.setdefault(event, []) handlers.append(callback) + pending = self._pending_pushes.pop(event, None) + if pending: + for payload in pending: + try: + callback(payload) + except Exception: + logger.exception( + "Error in handler for %s:%s (replayed)", self._topic, event + ) + def unsubscribe() -> None: handlers.remove(callback) @@ -225,13 +246,22 @@ def _on_message( elif event == "phx_error": self._handle_error(payload) else: - # User event — dispatch to handlers - handlers = self._event_handlers.get(event, []) - for handler in handlers: - try: - handler(payload) - except Exception: - logger.exception("Error in handler for %s:%s", self._topic, event) + # User event — dispatch to handlers, or buffer if none yet. + handlers = self._event_handlers.get(event) + if handlers: + for handler in handlers: + try: + handler(payload) + except Exception: + logger.exception("Error in handler for %s:%s", self._topic, event) + else: + buf = self._pending_pushes.setdefault(event, []) + if len(buf) < self._pending_pushes_cap: + buf.append(payload) + else: + # Drop the oldest to bound memory for orphan events. + buf.pop(0) + buf.append(payload) def _handle_reply(self, ref: str | None, payload: Any) -> None: if ref and ref in self._pending_replies: diff --git a/src/phx_channel/tests/test_unit.py b/src/phx_channel/tests/test_unit.py index ee80a08..a1d7da8 100644 --- a/src/phx_channel/tests/test_unit.py +++ b/src/phx_channel/tests/test_unit.py @@ -287,6 +287,35 @@ async def test_null_join_ref_accepted(): assert received == [{"data": "broadcast"}] +async def test_pushes_before_handler_are_replayed_on_registration(): + """ + Server `autoPush` can fire in response to a join frame and land on the + asyncio queue before the caller has registered a handler. Verify the + Channel buffers those pushes and replays them when on() is called. + """ + _, ch = await _joined_channel() + # Two pushes arrive with no handlers yet — both should be buffered. + ch._on_message(None, None, "new_entry", {"id": "a"}) + ch._on_message(None, None, "new_entry", {"id": "b"}) + received: list[dict] = [] + ch.on("new_entry", lambda p: received.append(p)) + assert received == [{"id": "a"}, {"id": "b"}] + # Subsequent pushes go straight through. + ch._on_message(None, None, "new_entry", {"id": "c"}) + assert received == [{"id": "a"}, {"id": "b"}, {"id": "c"}] + + +async def test_pending_push_buffer_is_bounded(): + _, ch = await _joined_channel() + ch._pending_pushes_cap = 3 + for i in range(5): + ch._on_message(None, None, "evt", {"i": i}) + received: list[dict] = [] + ch.on("evt", lambda p: received.append(p)) + # Oldest dropped once cap hit; newest kept in order. + assert received == [{"i": 2}, {"i": 3}, {"i": 4}] + + async def test_handler_error_does_not_crash_dispatch(): _, ch = await _joined_channel() received = [] From 12b90375dab8201631b13dd51e0a3b9c96059bc9 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Fri, 17 Apr 2026 15:19:07 -0700 Subject: [PATCH 2/2] fix(phx_channel): apply ruff format CI flagged the previous commit's edit for formatting; run ruff format so the check passes. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/phx_channel/channel.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/phx_channel/channel.py b/src/phx_channel/channel.py index e4747ae..e00f36c 100644 --- a/src/phx_channel/channel.py +++ b/src/phx_channel/channel.py @@ -217,9 +217,7 @@ def on(self, event: str, callback: Callable[..., Any]) -> Callable[[], None]: try: callback(payload) except Exception: - logger.exception( - "Error in handler for %s:%s (replayed)", self._topic, event - ) + logger.exception("Error in handler for %s:%s (replayed)", self._topic, event) def unsubscribe() -> None: handlers.remove(callback)