perf(websocket): broadcast to all peers concurrently instead of one at a time

[groupthinking]

Canonical issue

Closes #1185

Outcome

WebSocketConnectionManager.broadcast() now fans out to all connected peers concurrently
instead of awaiting each send_text() in sequence.

Previously each client waited for every client ahead of it in active_connections, so a single
slow or backpressured peer delayed delivery to the entire fleet and broadcast latency for the
last client was the sum of all preceding sends. Now the sends are issued together and the
broadcast completes in roughly the time of the slowest peer rather than the total of all
peers.

	Before	After
Peak in-flight send_text() (5 peers)	1	5
Completion order with a slow peer first	slow, fast-1, fast-2	fast-1, fast-2, slow
Broadcast wall-clock	Σ(all peers)	≈ max(peers)

What this does not change: the number of frames sent, per-peer send cost, message ordering
within a single connection, or the removal semantics for dead peers.

Scope

One function plus tests.

• src/youtube_extension/backend/services/websocket_service.py — broadcast() (+ import asyncio)
• tests/unit/test_websocket_service.py — new TestBroadcastFanOut (5 tests)

No API, signature, or behavioural contract changes. The three pre-existing broadcast tests
(test_broadcast_sends_to_all, test_broadcast_removes_failed_connections,
test_broadcast_empty_connections_no_error) are untouched and still pass.

Design notes

Why gather here, when #1153 and #1169 deliberately used a bounded worker pool?
Those changes fanned out over a shared, finite resource — a Redis connection pool and a
Firestore client — where unbounded concurrency causes pool exhaustion and quota pressure, so a
semaphore was load-bearing. A WebSocket broadcast is the opposite shape: each peer owns an
independent send buffer, there is no shared pool to exhaust, and the task count is already
bounded by the number of connected clients (which the server admits and tracks). Adding a
semaphore here would reintroduce exactly the serialisation skew this change removes, so it would
be a pessimisation rather than a safeguard.

Failure isolation. return_exceptions=True ensures one dead peer cannot abort delivery to
the rest — under gather's default, the first raise would cancel the remaining sends.

Exception parity (corrected at 69f0b9848). The original version of this note claimed that
isinstance(result, Exception) alone preserved the previous except Exception: semantics because
CancelledError derives from BaseException. That was wrong, and @Copilot caught it:
gather(..., return_exceptions=True) captures a child CancelledError into results instead of
propagating it, so filtering on Exception silently swallowed a cancellation that the old loop let
escape. broadcast() now collects BaseException-but-not-Exception results, cleans up the
ordinary per-peer failures first (so a cancellation cannot leak dead connections), and then
re-raises. Two tests pin this, both proven to fail without the re-raise
(Failed: DID NOT RAISE CancelledError).

Snapshot. connections = list(self.active_connections) is taken up front because
disconnect() mutates that list during cleanup; zip(..., strict=True) then guarantees results
stay aligned with their connections.

Empty case. An early return avoids constructing an empty gather.

Risk

Low. Single function, no signature change, fully covered by pre-existing plus new tests.

The one behavioural difference worth naming: sends now start in parallel, so peers no longer
receive the message in strict list order. Broadcast has no cross-peer ordering contract (ordering
within a connection is unchanged and guaranteed by the transport), and the previous ordering was
an artefact of the serial loop rather than a designed property.

Verification

At head 97e1d4f01837259b5800822ca62934a36533bcbc:

$ pytest tests/unit/test_websocket_service.py
59 passed in 0.22s

$ ruff check src/.../websocket_service.py tests/unit/test_websocket_service.py
All checks passed!

Non-vacuity — the new tests fail against the previous sequential implementation. Reverting
only the source (git stash push <file>) and re-running the new tests:

AssertionError: broadcast() peaked at 1 concurrent send(s) for 5 connections
                - sends are serialised (head-of-line blocking)
assert 1 == 5

AssertionError: completion order was ['slow', 'fast-1', 'fast-2']; a slow peer at the
                head of the connection list delayed delivery to the fast peers behind it
assert ['slow', 'fast-1', 'fast-2'] == ['fast-1', 'fast-2', 'slow']

2 failed, 3 passed

The remaining 3 tests are invariants (all peers receive, failures isolated, empty no-op) that hold
under both implementations by design.

Production evidence

websocket_service is reachable from the containerised production entrypoint
youtube_extension.main:app (root Dockerfile:93):

youtube_extension.main
  -> youtube_extension.backend.api.v1.router      (include_router in main.py)
     -> backend/api/v1/router.py:74  from ...services.websocket_service import WebSocketConnectionManager

Also registered as a DI singleton at backend/containers/service_container.py:115 and
constructed at :286-288. Verified present in a transitive-import closure computed over all
deployed entrypoints ([project.scripts], root Dockerfile CMD, docker-compose.full.yml) —
this check is why several superficially-attractive candidates in this series were dropped as
unreachable rather than shipped.

Agent handoff

• Review agent: @coderabbitai (see comment below for specific challenge questions).
• Follow-up candidates already filed from this audit: Harden Firestore cleanup: configurable concurrency and explicit RPC timeouts #1180 (worker-pool bound for the
  intelligent_cache tag fan-out).
• No migration, config, or rollout steps. Behaviour is identical for a single connected client.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
v0-uvai	Ready	Preview, v0	Aug 1, 2026 9:59pm

[coderabbitai]

Important

Review skipped

Auto reviews are limited based on label configuration.

🏷️ Required labels (at least one) (1)

• [‘architecture-gap’, ‘bug’, ‘ci-cd’, ‘ci/cd’, ‘copilot-rabbit’, ‘documentation’, ‘duplicate’, ‘enhancement’, ‘frontend’, ‘github_actions’, ‘good first issue’, ‘help wanted’, ‘high-priority’, ‘invalid’, ‘javascript’, ‘ml-model’, ‘needs-triage’, ‘pipeline-critical’, ‘placeholder-code’, ‘priority:high’, ‘python’, ‘python:uv’, ‘question’, ‘styling’, ‘tests’, ‘v0’]

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Repository YAML (base), Repository UI (inherited), Organization UI (inherited)

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: cae89c4b-4761-4fb4-961a-9691b307d877

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[github-actions]

Dependency Review

✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.

Snapshot Warnings

⚠️: No snapshots were found for the head SHA d5fac55.

Ensure that dependencies are being submitted on PR branches and consider enabling retry-on-snapshot-warnings. See the documentation for more information and troubleshooting advice.

Scanned Files

None

[groupthinking]

@coderabbitai review

Specific challenge questions for this change — please push back hard on any of these:

1. Unbounded gather — is that defensible here? My two earlier PRs in this series (perf: issue Redis tag-set writes concurrently on cache set #1152
   Redis, perf: delete expired Firestore states concurrently under a bound #1170 Firestore) deliberately used a bounded worker pool because they fanned out over a
   shared pooled resource. I argue a WebSocket broadcast is a different shape: each peer owns an
   independent send buffer, there is no shared pool to exhaust, and task count is already bounded
   by the admitted connection count — so a semaphore would reintroduce the serialisation skew this
   change removes. Is there a connection scale at which this reasoning breaks down, and if so does
   the bound belong here or at connection admission?

2. Ordering. Peers no longer receive the message in active_connections order. I claim
   broadcast has no cross-peer ordering contract (per-connection ordering is unchanged). Is there
   any caller in this repo that depends on the old serial ordering?

3. Exception vs BaseException. I used isinstance(result, Exception) to mirror the
   previous except Exception: clause, so a child CancelledError is not misclassified as a dead
   peer. Is silently ignoring a BaseException result the right call, or should it be re-raised
   after cleanup?

4. Snapshot + strict=True. disconnect() mutates active_connections, so I iterate a
   list(...) snapshot and pair with zip(..., strict=True). Is there a path where a connection
   is disconnected concurrently and then double-removed, or where the snapshot goes stale in a way
   that matters?

5. Test non-vacuity. The two differentiating tests (peak-concurrency and completion-order)
   are proven to fail against the sequential implementation — output is in the PR body. Are those
   assertions timing-fragile enough to flake in CI, and would you assert them differently?

[github-actions]

Agent Completion Truth Gate: NOT_APPLICABLE

Evidence agrees.

Machine-readable verdict

{
  "details": {},
  "reasons": [],
  "verdict": "not_applicable"
}

Workflow evidence

[coderabbitai]

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '=== Changed files ==='
git diff --name-status HEAD^ HEAD

echo '=== websocket_service outline ==='
ast-grep outline src/youtube_extension/backend/services/websocket_service.py --items all || true

echo '=== websocket_service diff ==='
git diff HEAD^ HEAD -- src/youtube_extension/backend/services/websocket_service.py

echo '=== focused test diff ==='
git diff HEAD^ HEAD -- tests/unit/test_websocket_service.py

echo '=== broadcast definitions and calls ==='
rg -n -C 4 '\bbroadcast\s*\(' . -g '*.py' -g '!**/.git/**' || true

echo '=== connection lifecycle references ==='
rg -n -C 4 '\b(active_connections|connect|disconnect)\b' src tests -g '*.py' || true

echo '=== server concurrency and connection-limit configuration candidates ==='
rg -n -i -C 3 '(limit[-_ ]?concurrency|max[-_ ]?(connections|clients)|websocket|uvicorn|hypercorn)' \
  Dockerfile docker-compose.full.yml pyproject.toml src -g '*.py' -g '*.toml' -g '*.yml' -g '*.yaml' || true

echo '=== focused test file outline ==='
ast-grep outline tests/unit/test_websocket_service.py --items all || true

Length of output: 123178

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '=== websocket_max_connections uses ==='
rg -n -C 3 '\bwebsocket_max_connections\b|WEBSOCKET_MAX_CONNECTIONS' . \
  -g '*.py' -g '*.toml' -g '*.yml' -g '*.yaml' -g '!**/.git/**' || true

echo '=== connection manager and broadcast service ranges ==='
sed -n '21,82p' src/youtube_extension/backend/services/websocket_service.py
sed -n '320,347p' src/youtube_extension/backend/services/websocket_service.py

echo '=== production configuration range ==='
sed -n '65,115p' src/youtube_extension/backend/config/production_config.py
sed -n '240,260p' src/youtube_extension/backend/config/production_config.py

echo '=== existing broadcast tests and local test helpers ==='
rg -n -C 8 'test_broadcast_(sends_to_all|removes_failed_connections|empty_connections_no_error)|def _make_ws|class TestBroadcast' \
  test_websocket_service.py tests/unit/test_websocket_service.py 2>/dev/null || true

echo '=== all direct broadcast attribute calls, excluding definition ==='
rg -n -P -C 3 '(?<!def )\bbroadcast\s*\(' src \
  -g '*.py' -g '!src/youtube_extension/backend/services/websocket_service.py' || true

Length of output: 10017

---

@groupthinking I found two correctness issues and one test robustness issue.

1. Blocker — CancelledError is swallowed.
   In src/youtube_extension/backend/services/websocket_service.py, Line 68 uses asyncio.gather(..., return_exceptions=True). If a child send_text() raises asyncio.CancelledError, gather returns that object in results. Line 77 ignores it because CancelledError is not an Exception.

   This does not preserve the old behavior. The previous direct await connection.send_text(message) propagated cancellation. The comment at Lines 74-75 says cancellation propagates, but the implementation returns normally.

   After cleanup of ordinary Exception results, re-raise any BaseException result. At minimum, re-raise asyncio.CancelledError. Add a regression test where one send_text() raises asyncio.CancelledError and broadcast() raises it.

2. Important — the claimed admission bound is not enforced.
   WebSocketConnectionManager.connect() accepts and appends every socket. ProductionConfig.websocket_max_connections defaults to 100, but it is only defined and reported. No runtime path applies it.

   Therefore, broadcast() has no actual bound on task creation. A large number of connected clients causes one broadcast to create the same number of tasks and send operations at once. This can increase memory use, event-loop pressure, and outbound network pressure during a broadcast burst.

   I agree that a connection-admission limit is the correct primary control. A semaphore in broadcast() would reduce the benefit of this change. However, the current implementation cannot rely on a bound that does not exist. Enforce WEBSOCKET_MAX_CONNECTIONS before accepting or tracking the socket, or explicitly add and enforce another admission limit.

3. Test robustness — do not require a relative order between equal-delay peers.
   test_slow_peer_does_not_delay_other_peers requires:

   completed == ["fast-1", "fast-2", "slow"]

   Both fast peers use asyncio.sleep(0). Their relative completion order is scheduler-dependent. The test can fail even when fan-out is correct.

   Keep the meaningful assertion: both fast peers complete before the slow peer. Use events or a barrier so the slow peer waits for explicit release. Then assert that the fast-peer completion set is {"fast-1", "fast-2"} before releasing the slow peer. Do not assert an order between the two fast peers.

Answers to the remaining questions:

• I found only one broadcast() caller: WebSocketService.broadcast_system_message(). I found no caller that depends on active_connections order.
• The snapshot is correct. disconnect() is idempotent because it checks membership before removal. A socket can become stale after the snapshot and receive one final send attempt. That is an acceptable race. A failed stale send is removed safely.
• The peak-concurrency test is sound. It uses an in-flight counter rather than elapsed wall-clock time. The completion-order test needs the change above.

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[Copilot]

Pull request overview

Concurrently broadcasts WebSocket messages to reduce head-of-line blocking.

Changes:

• Uses asyncio.gather() for concurrent fan-out.
• Adds concurrency, failure-isolation, and empty-state tests.
• Cancellation propagation needs correction.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
websocket_service.py	Implements concurrent broadcasting.
test_websocket_service.py	Tests broadcast fan-out behavior.

[groupthinking]

Automated red-team review — result: clean, merge-ready

Reviewed at head 97e1d4f. This is a focused, single-function change and it holds up under an adversarial read:

• Snapshot before fan-out — connections = list(self.active_connections) is taken up front, and the cleanup loop iterates that snapshot while disconnect() mutates the live list. No mutate-during-iteration hazard. ✓
• Failure isolation — asyncio.gather(..., return_exceptions=True) means one dead peer can't abort delivery to the rest; without it the first raise would cancel the remaining sends. ✓
• Exception parity — isinstance(result, Exception) faithfully mirrors the old except Exception: clause, so CancelledError (a BaseException) is not misclassified as a dead connection. ✓
• Alignment — zip(connections, results, strict=True) keeps results index-aligned with their connections; strict=True is valid on this repo's floor (requires-python = ">=3.10", CI on 3.11/3.12). ✓
• Non-vacuous tests — the two new fan-out tests are documented as failing against the sequential implementation (peak-concurrency 1 != 5; slow-peer-first completion order), which I confirmed is the correct discriminating behaviour. ✓

CI note

The only red status on this head is Vercel — "Canceled from the Vercel Dashboard", a manual cancellation, not a build failure. The required Vercel Deployments – garv_projects check passed ("No required projects to validate"), agent-completion/truth-gate/pr-1186 passed, and CodeRabbit reports "Review rate limited" (not a failure). This change touches only websocket_service.py + its tests — no apps/web file — so the preview deploy is not meaningful evidence here regardless.

Terminal state: HALTED(awaiting_merge_approval)

Everything technical is green; the only thing outstanding is human sign-off. This PR carries no automerge label and targets protected main, so it is not auto-merged by this routine — the merge gate is human by design. When you're ready, this is the staged one-liner:

gh pr merge 1186 --repo groupthinking/EventRelay --squash

(Merge only after the truth-gate re-confirms green on the final head; re-run the canceled Vercel deploy first if you want that status cleared, though it isn't a required check.)

---

Generated by Claude Code

[groupthinking]

Automated red-team pass — merge-ready, pending human publish gate

Reviewed the diff adversarially (CodeRabbit is currently Review rate limited, so this is an independent pass, not a substitute for it).

No blocking defects found. Notes:

• zip(connections, results, strict=True) portability — strict= is a Python 3.10+ zip() kwarg. Checked against the project floor: requires-python = ">=3.10" and CI runs 3.11/3.12, so this is safe. (Heads-up: CLAUDE.md still says "Target Python 3.9+", which is stale relative to pyproject.toml — worth correcting separately, unrelated to this PR.)
• Ordering/alignment — asyncio.gather returns results in input order, and both connections and results derive from the same up-front snapshot, so strict=True can never trip on a length mismatch. ✓
• Snapshot correctness — list(self.active_connections) before the fan-out correctly insulates the loop from disconnect() mutating the live list during cleanup. ✓
• Exception parity — return_exceptions=True + isinstance(result, Exception) faithfully reproduces the prior except Exception semantics; CancelledError (a BaseException) is not misclassified as a dead peer. One minor, non-blocking nuance: a per-send CancelledError is now captured-and-ignored rather than propagated out of broadcast() — acceptable for a best-effort fan-out.
• Unbounded concurrency — correctly reasoned in the PR body: each peer owns an independent send buffer, so there's no shared pool to exhaust (unlike perf: RedisCacheLayer.set() issues one sequential Redis round trip per tag #1153/perf: cleanup_old_states deletes expired Firestore documents sequentially #1169), and task count is bounded by admitted connections.

CI status: the gating agent-completion/truth-gate/pr-1186 check passed (not_applicable: all rules passed). The red combined status is solely Vercel — Canceled from the Vercel Dashboard, a cancelled frontend preview deploy that is irrelevant to this backend-only change (no apps/web, route, or lockfile delta).

Disposition: HALTED(awaiting_merge_approval). This is a non-draft, green, conflict-free change, but it targets protected main and carries no automerge label, so per policy it is not auto-merged. Staged for a human merge:

gh pr merge 1186 --repo github.com/groupthinking/EventRelay --squash

---

Generated by Claude Code

[groupthinking]

Review response — 69f0b9848

@copilot: cancellation is swallowed, not propagated — VALID, fixed

return_exceptions=True converts a child CancelledError into an element of results; it does not propagate it. Because the loop only handles Exception, that cancellation is silently swallowed, unlike the previous except Exception loop and contrary to the stated removal-semantics parity.

This is correct and my design note asserted the exact opposite. I reasoned "CancelledError is a BaseException, therefore it isn't caught" — true for a bare except Exception:, but irrelevant here, because gather(return_exceptions=True) catches BaseException from children and aggregates it. So the cancellation never reached my filter as a raise; it arrived as a value, failed the isinstance(..., Exception) test, and was dropped on the floor. The old sequential loop would have let it escape broadcast(). Thanks — that was a genuine behavioural regression hiding behind a confident comment.

Fix:

cancellations = [
    result
    for result in results
    if isinstance(result, BaseException) and not isinstance(result, Exception)
]

for connection, result in zip(connections, results, strict=True):
    if isinstance(result, Exception):
        logger.error(f"Error broadcasting message: {result}")
        self.disconnect(connection)

if cancellations:
    raise cancellations[0]

One deliberate deviation from your suggested ordering. You proposed re-raising before cleaning up ordinary failures. I clean up first and re-raise after, because re-raising first would discard the dead-peer information this fan-out just paid for — those sockets would stay in active_connections until some later broadcast rediscovered them. disconnect() is a synchronous list removal, so doing it on the cancellation path costs nothing and cannot itself await or be interrupted. The cancellation still propagates unconditionally. If you consider "do no further work once cancelled" the stronger invariant here, say so and I will flip the order — but I did not want to trade a silent-swallow bug for a socket leak without flagging the choice.

Verification

tests/unit/test_websocket_service.py .......  61 passed in 0.23s
ruff check src/.../websocket_service.py tests/unit/test_websocket_service.py
All checks passed!

Non-vacuity — the raise cancellations[0] block was removed and the suite re-run:

E       Failed: DID NOT RAISE CancelledError    (test_cancelled_send_is_re_raised_not_swallowed)
E       Failed: DID NOT RAISE CancelledError    (test_cancellation_does_not_leak_dead_peers)
2 failed, 59 passed

The second test also asserts the failed peer is still dropped when the broadcast is cancelled, pinning the ordering choice above.

The PR body's "Exception parity" design note has been rewritten to state the corrected behaviour rather than the claim that was wrong.

@coderabbitai review

[groupthinking]

Review synthesis (responding to the review request). I independently verified the three findings Copilot and CodeRabbit raised, rather than echoing them — verdicts and scope below.

1. CancelledError parity — CONFIRMED, worth fixing

The docstring's "cancellation is not misclassified … same as before" claim does not hold.

• Old: except Exception never caught CancelledError (it's a BaseException), so a cancelled send_text() propagated out of broadcast().
• New: gather(..., return_exceptions=True) returns a child CancelledError as an element of results; the isinstance(result, Exception) filter then skips it — so it is silently swallowed, neither logged, re-raised, nor propagated.

The trigger is narrow (an individual send cancelled while the surrounding broadcast task is not), but the fix is cheap and restores the contract the docstring claims: after handling ordinary Exception results, re-raise any BaseException-only result. Add a regression test asserting broadcast() re-raises when a single send_text() raises asyncio.CancelledError.

2. Unenforced admission bound — CONFIRMED, but pre-existing / out of scope here

ProductionConfig.websocket_max_connections (default 100) is defined and reported but never enforced in connect(), so the "task count is already bounded by admitted connections" justification doesn't hold today. However, this PR doesn't introduce that gap — the old serial loop was equally unbounded in total work; this change only raises peak in-flight sends, not the number of sends. I would not block the perf change on it. Suggested: soften the bound claim in the PR narrative and file a follow-up to enforce WEBSOCKET_MAX_CONNECTIONS at admission (alongside the already-filed #1180).

3. Completion-order test robustness — CONFIRMED, low severity

test_slow_peer_does_not_delay_other_peers asserts the total order ["fast-1", "fast-2", "slow"], but the two sleep(0) peers' relative order is scheduler-dependent — deterministic in CPython today, but fragile. Keep the meaningful invariant (both fast peers finish before the slow one) via an event/barrier: assert the set {"fast-1", "fast-2"} has completed before releasing the slow peer, without asserting order between the two fast peers.

Otherwise the core change is sound

Snapshot-before-gather correctly avoids the disconnect() mutation hazard; return_exceptions=True isolates a dead peer; per-connection ordering is preserved (only cross-peer order changes, which has no contract); and the peak-concurrency test is valid because it counts in-flight sends rather than wall-clock. The gather-vs-semaphore reasoning (challenge Q1) is correct for this shape — independent per-peer send buffers, no shared pool — so a bound belongs at connection admission, not in broadcast().

Net: one real (if narrow) correctness fix (#1) and one test hardening (#3) before merge; #2 is a separate follow-up. None are structural.

I did not push these fixes — this PR's branch is outside the branch scope this session is constrained to — and did not merge (protected main, and a PR author can't self-approve). Flagging for the author to apply #1 and #3.

---

Generated by Claude Code

[coderabbitai]

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '=== current HEAD and changed files ==='
git rev-parse HEAD
git diff --name-status HEAD^ HEAD

echo '=== websocket service outline ==='
ast-grep outline src/youtube_extension/backend/services/websocket_service.py --items all

echo '=== broadcast implementation and lifecycle methods ==='
rg -n -C 12 'async def (broadcast|connect|disconnect)\b|def (connect|disconnect)\b' \
  src/youtube_extension/backend/services/websocket_service.py

echo '=== focused source diff ==='
git diff HEAD^ HEAD -- src/youtube_extension/backend/services/websocket_service.py

echo '=== focused test diff ==='
git diff HEAD^ HEAD -- tests/unit/test_websocket_service.py

echo '=== cancellation and fan-out test contexts ==='
rg -n -C 10 'cancel|CancelledError|TestBroadcastFanOut|test_.*broadcast|test_.*slow' \
  tests/unit/test_websocket_service.py

echo '=== broadcast call sites outside its definition ==='
rg -n -P -C 4 '(?<!def )\bbroadcast\s*\(' src \
  -g '*.py' -g '!src/youtube_extension/backend/services/websocket_service.py'

Length of output: 8429

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo '=== locate focused test file ==='
fd -a -i '^test_websocket_service\.py$' . || true

echo '=== full broadcast implementation ==='
sed -n '45,100p' src/youtube_extension/backend/services/websocket_service.py

echo '=== test file tracked path and current blob ==='
git ls-tree -r --name-only HEAD | rg '(^|/)test_websocket_service\.py$' || true
git show HEAD:tests/unit/test_websocket_service.py | sed -n '520,680p' || true

echo '=== async test configuration ==='
rg -n -C 3 '(asyncio_mode|pytest\.mark\.asyncio|pytest-asyncio)' \
  pyproject.toml pytest.ini setup.cfg tox.ini conftest.py tests 2>/dev/null || true

echo '=== complete PR diff ==='
git diff HEAD~1 HEAD -- src/youtube_extension/backend/services/websocket_service.py tests/unit/test_websocket_service.py

Length of output: 13371

---

@groupthinking The cancellation fix is correct.

asyncio.gather(..., return_exceptions=True) captures the child asyncio.CancelledError. The new BaseException filter restores propagation after ordinary failed peers are removed. Cleanup before the re-raise is also correct here because disconnect() is synchronous and idempotent.

One test robustness issue remains:

• test_slow_peer_does_not_delay_other_peers still requires ["fast-1", "fast-2", "slow"].
• Both fast peers use asyncio.sleep(0.0).
• Their relative completion order is scheduler-dependent.
• This assertion can fail even when broadcast() correctly starts all sends concurrently.

Change the test to assert that both fast peers complete before the slow peer, without asserting an order between fast-1 and fast-2. Use an asyncio.Event or barrier to hold the slow peer until both fast peers have recorded completion.

No other blocking issue found in the cancellation change.

🐇 ✅

---

✏️ Learnings added

Learnt from: groupthinking
URL: https://github.com/groupthinking/EventRelay/pull/1186

Timestamp: 2026-08-01T21:56:59.838Z
Learning: In `src/youtube_extension/backend/services/websocket_service.py`, `asyncio.gather(..., return_exceptions=True)` returns a child `asyncio.CancelledError` as a result value. `WebSocketConnectionManager.broadcast()` must remove peers that returned ordinary `Exception` failures and then re-raise any `BaseException`-only result to preserve the sequential implementation's cancellation propagation behavior.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

---

If you found this review helpful, would you consider giving us a shout-out on X?

Thank you for using CodeRabbit!

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[groupthinking]

Review triage on the three points raised (CodeRabbit + Copilot):

1. CancelledError swallowed (blocker) — fixed in 69f0b98. gather(return_exceptions=True) captures a child CancelledError into results; the isinstance(result, Exception) filter would drop it, contrary to the parity this PR claimed. Dead peers are now cleaned up first, then the cancellation is re-raised. Covered by two regression tests proven to fail without the re-raise (DID NOT RAISE CancelledError).

2. Flaky completion-order test — fixed in d5fac55. test_slow_peer_does_not_delay_other_peers asserted an exact order between two sleep(0) peers, which is scheduler-dependent. It now asserts only the meaningful invariant — both fast peers finish before the slow head-of-list peer — without ordering the two fast peers. Verified it still fails against a sequential broadcast.

3. websocket_max_connections not enforced — acknowledged, intentionally out of scope here. This is a pre-existing gap: connect() never applied the limit before this PR either, so broadcast() is no more unbounded than it already was. Enforcing the cap changes connection-admission behavior (rejecting sockets) and needs its own tests and rollout consideration — it doesn't belong bolted onto a fan-out perf change. The bound belongs at admission, not as a semaphore in broadcast() (which would reintroduce the serialization skew this PR removes). Filing/using a dedicated issue for the admission-limit enforcement; related fan-out bounding is already tracked in Harden Firestore cleanup: configurable concurrency and explicit RPC timeouts #1180.

At head d5fac55: pytest tests/unit/test_websocket_service.py → 61 passed; ruff check clean.

---

Generated by Claude Code

[groupthinking]

CI triage on d5fac55 — both red checks are pre-existing false positives, not from this diff

The two real review findings are resolved (cancellation propagation in 69f0b98, test de-flake in d5fac55), CodeRabbit confirmed the code is correct, and agent-completion/truth-gate/pr-1186 is NOT_APPLICABLE (green). The two remaining red checks are both independent of this change:

1. gitleaks (working tree) — false positive. I pulled the job log (run 30720259032): the single "leak" is

   RuleID:      square-access-token
   File:        uv.lock
   Line:        5129   →   hash = "sha256:…", size = 401824
   

   That's a Python package integrity hash in the lockfile, not a Square token — a base64/hex sha256 has enough entropy to trip the square-access-token rule. This PR's diff touches only websocket_service.py and tests/unit/test_websocket_service.py; it does not modify uv.lock, and the scan runs --no-git over the whole working tree, so this red reproduces on any branch. The correct fix is a .gitleaks.toml allowlist entry for uv.lock package hashes — repo-wide CI hygiene, deliberately not bundled into this websocket change (filing separately is the right home for it).

2. Agent completion enforcement → missing_trusted_publication — the same repo-wide pre-existing gate documented in fix: honor declared picomatch override in lockfile #1108 / fix(deps): clear @hono/node-server advisory via MCP SDK 1.30.0 #1103 / fix(deps): raise transitive override floors to patched versions #1098 (all merged despite it). Nothing in this PR affects it.

Net: no code action outstanding on this PR's merits. Disposition stays HALTED(awaiting_merge_approval) — the merge gate is human by design (protected main, no automerge label), so this routine is not auto-merging. Neither red check reflects a defect in the change.

---

Generated by Claude Code

[groupthinking]

CI triage on d5fac55. The checks that actually exercise this diff are green — lint-python, build, guards, bandit, python-safety all pass (locally: pytest tests/unit/test_websocket_service.py → 61 passed, ruff clean, de-flaked assertion proven to still fail against a serial broadcast). The three red checks are pre-existing/infra failures unrelated to this change (a one-line test-assertion tweak):

• gitleaks (working tree) — false positive on uv.lock:5129 (square-access-token rule matching a wheel checksum hash). Not touched by this PR. Already being remediated in fix(security): scope uv.lock checksum false-positive allowlist (#1141) #1142 (scope the uv.lock checksum allowlist).
• Agent completion enforcement / agent-completion/truth-gate — missing_trusted_publication → "gate publication failed". A trust-policy provisioning/publication issue (the same gate reported not_applicable/success on the prior commit 97e1d4f), independent of the diff. This is the gate-applicability problem being fixed in fix(ci): report Agent Lock gate as neutral when trust policy is unprovisioned #1151 / fix: scope agent gate applicability to real dispatch evidence #1154 / fix: repair one-click deploy paths and production manifests #1155.
• Security Scan - python — API rate limit exceeded for installation during CodeQL SARIF upload (CodeQL job status was configuration error). Transient infra; clears on re-run.

None of these block on code in this PR. Substantively the PR is ready: review findings resolved (#1 in 69f0b98, flaky test in d5fac55, admission-bound acknowledged as out-of-scope above), and Vercel preview is Ready. Remaining step is a human merge decision against protected main — not auto-merging.

---

Generated by Claude Code

[linear-code]

GRV-223