From 9738ebb6147ed0856d1c0127ca96dcf0034fda83 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Sun, 5 Jul 2026 19:52:16 -0700 Subject: [PATCH] fix(tests): sync on subscriber count in multitab SSE test (#5628) (#5660) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_same_stream_in_two_tabs_receives_identical_token_sequence started both SSE-handler threads then immediately put_nowait'd the tokens + stream_end without waiting for both threads to subscribe. StreamChannel.put_nowait() buffers events only while there are zero subscribers and clears that buffer the moment the first subscriber attaches and broadcasts live — so producing events in the window after tab A subscribes but before tab B does orphans tab B (subscribes to a cleared buffer, never sees stream_end, hangs -> thread.join timeout -> 'every tab should finish' fails). ~20% timing flake on slow CI runners (3.11 shard-4). Wait until len(stream._subscribers) == len(handlers) (bounded 5s) before producing events, so delivery is deterministic regardless of thread-start timing. Test-only; no production change (buffer-then-broadcast is correct for the real reconnect case). Verified: original flakes 2/30 under CPU contention; fixed 30/30. Fixes #5628 Co-authored-by: nesquena-hermes --- tests/test_issue_1584_multitab_sse.py | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/tests/test_issue_1584_multitab_sse.py b/tests/test_issue_1584_multitab_sse.py index 8341d317..925055a4 100644 --- a/tests/test_issue_1584_multitab_sse.py +++ b/tests/test_issue_1584_multitab_sse.py @@ -1,5 +1,6 @@ import io import threading +import time from types import SimpleNamespace from api.config import STREAMS, STREAMS_LOCK, create_stream_channel @@ -64,6 +65,28 @@ def test_same_stream_in_two_tabs_receives_identical_token_sequence(): for thread in threads: thread.start() + # Wait until BOTH tabs have actually subscribed before producing any + # events. StreamChannel.put_nowait() buffers events in _offline_buffer + # only while there are ZERO subscribers, and clears that buffer the + # moment the first subscriber attaches and it broadcasts live. If we + # put tokens in the window after tab A subscribes but before tab B + # does, tab B subscribes to an already-cleared buffer, never receives + # stream_end, and hangs -> thread.join times out (the ~20% timing flake + # on slow CI runners, issue #5628). Synchronize on the real subscriber + # count so the test is deterministic regardless of thread-start timing. + deadline = time.monotonic() + 5.0 + subscribed = 0 + while time.monotonic() < deadline: + with stream._lock: + subscribed = len(stream._subscribers) + if subscribed >= len(handlers): + break + time.sleep(0.005) + else: + raise AssertionError( + f"only {subscribed}/{len(handlers)} tabs subscribed before timeout" + ) + stream.put_nowait(("token", {"text": "H"})) stream.put_nowait(("token", {"text": "allo"})) stream.put_nowait(("stream_end", {"status": "done"}))