From 179c149c46a69b976280b20be861ed9a9f0ba6a8 Mon Sep 17 00:00:00 2001 From: Yi Ding Date: Thu, 6 Aug 2026 08:18:00 -0700 Subject: [PATCH 1/2] server: report WebRTC cleanup failures --- .../07-transports-and-production.md | 2 + src/easycat/server/webrtc_routes.py | 61 ++++++++++++++++--- tests/ratchets/source-baseline.json | 7 +-- tests/server/test_capacity_gate_drain.py | 40 ++++++++++++ 4 files changed, 97 insertions(+), 13 deletions(-) diff --git a/docs/development/07-transports-and-production.md b/docs/development/07-transports-and-production.md index c01d54ade..015ce79fd 100644 --- a/docs/development/07-transports-and-production.md +++ b/docs/development/07-transports-and-production.md @@ -227,6 +227,8 @@ WebRTC has two layers: stateless/per-server route handling, admission, and per-offer transport construction. Each accepted offer also installs one named runtime cleanup task that waits for peer closure and releases its Session and capacity slot. + Shutdown cancels those waiters, runs forced finalizers in a separate named + cohort, and reports unexpected results with their task identities. Shared config/stats/health/CORS/root behavior lives once in [`server/_webrtc_handlers.py`](../../src/easycat/server/_webrtc_handlers.py). diff --git a/src/easycat/server/webrtc_routes.py b/src/easycat/server/webrtc_routes.py index 073717b7a..cf3ff1712 100644 --- a/src/easycat/server/webrtc_routes.py +++ b/src/easycat/server/webrtc_routes.py @@ -83,6 +83,8 @@ _OFFER_CLEANUP_TASK = "webrtc_offer_cleanup" _OFFER_CLEANUP_COHORT = "webrtc-offer-cleanup" +_FORCE_CLEANUP_TASK = "webrtc_force_cleanup" +_FORCE_CLEANUP_COHORT = "webrtc-force-cleanup" # Per-connection factory seam (NO ``ConnectionContext`` type): a per-transport # ``Callable[[WebRTCTransport], EasyConfig | Session]``. @@ -188,6 +190,14 @@ def __init__( failure_message="WebRTC offer cleanup task failed", drop_if_closed=False, ) + self._force_cleanup_task_scope = RuntimeTaskScope( + owner_label="webrtc-routes-force-cleanup", + member_name=_FORCE_CLEANUP_TASK, + cohort=_FORCE_CLEANUP_COHORT, + logger=logger, + failure_message="WebRTC forced cleanup task failed", + drop_if_closed=False, + ) self._cleanup_task_keys: dict[asyncio.Task[Any], int] = {} self._released_cleanup_keys: set[int] = set() # aiohttp.web, resolved lazily inside ``register``. @@ -561,21 +571,54 @@ async def cancel_cleanup_tasks(self, *, timeout_s: float | None = None) -> None: for task, _key in pending: task.cancel() if pending: - finalizers = [ - asyncio.create_task(self._finalize_session_cleanup(key, force=True)) - for _task, key in pending - if key is not None - ] + finalizers: list[asyncio.Task[Any]] = [] + for _task, key in pending: + if key is None: + continue + finalizer = self._force_cleanup_task_scope.create_task( + self._finalize_session_cleanup(key, force=True), + task_name=f"easycat-webrtc-force-cleanup-{key}", + ) + assert finalizer is not None + finalizers.append(finalizer) + cleanup_tasks = [*(task for task, _key in pending), *finalizers] cleanup = asyncio.gather( - *(task for task, _key in pending), - *finalizers, + *cleanup_tasks, return_exceptions=True, ) + results: list[Any | BaseException] | None if timeout_s is None: - await cleanup + results = await cleanup else: - await _await_with_hard_timeout(cleanup, timeout_s=timeout_s) + completed = await _await_with_hard_timeout(cleanup, timeout_s=timeout_s) + results = cleanup.result() if completed else None + if results is not None: + self._report_cleanup_results( + cleanup_tasks, + results, + explicitly_cancelled={task for task, _key in pending}, + ) await self._cleanup_task_scope.release_standalone_if_empty() + await self._force_cleanup_task_scope.release_standalone_if_empty() + + def _report_cleanup_results( + self, + tasks: list[asyncio.Task[Any]], + results: list[Any], + *, + explicitly_cancelled: set[asyncio.Task[Any]], + ) -> None: + """Log unexpected cleanup failures with the owning task's identity.""" + for task, result in zip(tasks, results, strict=True): + if not isinstance(result, BaseException): + continue + if isinstance(result, asyncio.CancelledError) and task in explicitly_cancelled: + continue + logger.error( + "WebRTC cleanup task %s failed", + task.get_name(), + exc_info=result, + ) async def _stop_managed_session(self, key: int, force: bool) -> None: """Route drain teardown through the manager's keyed stop ownership.""" diff --git a/tests/ratchets/source-baseline.json b/tests/ratchets/source-baseline.json index 042fd61d2..3ff29f4ee 100644 --- a/tests/ratchets/source-baseline.json +++ b/tests/ratchets/source-baseline.json @@ -1,12 +1,12 @@ { "version": 1, - "rationale": "WebRTC per-offer cleanup workers now use named RuntimeTaskScope ownership.", + "rationale": "WebRTC forced cleanup finalizers now use named ownership and report unexpected results.", "counts": { "cancelled_error_handler": 100, "epoch_field": 17, "gather_return_exceptions": 22, "module_task_set": 1, - "raw_task_spawn": 34, + "raw_task_spawn": 33, "shield_loop": 4, "task_cancelling": 51 }, @@ -136,7 +136,7 @@ "gather_return_exceptions\tserver/transports.py\tclose_websocket_connections\tasyncio.gather(return_exceptions=True)\t768ec4176f5a8ad9\t0", "gather_return_exceptions\tserver/voice_server.py\tVoiceServer._cancel_ws_handler_tasks\tasyncio.gather(return_exceptions=True)\t77ded9771f58812f\t0", "gather_return_exceptions\tserver/voice_server.py\tVoiceServer._close_active_ws_connections\tasyncio.gather(return_exceptions=True)\t4d4f60904d8d66b3\t0", - "gather_return_exceptions\tserver/webrtc_routes.py\tWebRTCRoutes.cancel_cleanup_tasks\tasyncio.gather(return_exceptions=True)\t69a335c95c0827de\t0", + "gather_return_exceptions\tserver/webrtc_routes.py\tWebRTCRoutes.cancel_cleanup_tasks\tasyncio.gather(return_exceptions=True)\t189be02028bb19e4\t0", "gather_return_exceptions\tsession/_tts_scheduler.py\tTTSScheduler.begin_synthesis_with_bot_start\tasyncio.gather(return_exceptions=True)\t42eb62b100557e55\t0", "gather_return_exceptions\tsession/_tts_scheduler.py\tTTSScheduler.begin_synthesis_with_bot_start\tasyncio.gather(return_exceptions=True)\t42eb62b100557e55\t1", "gather_return_exceptions\tsession/_tts_scheduler.py\tTTSScheduler.begin_synthesis_with_bot_start\tasyncio.gather(return_exceptions=True)\t42eb62b100557e55\t2", @@ -168,7 +168,6 @@ "raw_task_spawn\tserver/voice_server.py\tVoiceServer._close_active_ws_connections\tasyncio.create_task\tbcc4bc77941655eb\t0", "raw_task_spawn\tserver/voice_server.py\tVoiceServer._prepare_listener_cleanup_task\tasyncio.ensure_future\t5c7a78f6e7637ffa\t0", "raw_task_spawn\tserver/voice_server.py\tVoiceServer._stop_unlocked\tasyncio.create_task\t17a66ec385918dba\t0", - "raw_task_spawn\tserver/webrtc_routes.py\tWebRTCRoutes.cancel_cleanup_tasks\tasyncio.create_task\tf891ff9a3484306c\t0", "raw_task_spawn\tserver/webrtc_routes.py\t_shutdown_standalone_webrtc\tasyncio.create_task\tb2e5ccf8c0d214f5\t0", "raw_task_spawn\tsession/_journal_sink.py\tSessionJournalSink.append_record_async\tasyncio.create_task\t933bcec6345d252c\t0", "raw_task_spawn\tsession/_session.py\tSession._cut_off_turn_playback\tasyncio.create_task\t002fb7c8f4960eaf\t0", diff --git a/tests/server/test_capacity_gate_drain.py b/tests/server/test_capacity_gate_drain.py index 2194e65ea..c055af012 100644 --- a/tests/server/test_capacity_gate_drain.py +++ b/tests/server/test_capacity_gate_drain.py @@ -737,6 +737,46 @@ async def test_webrtc_cleanup_and_drain_share_one_force_escalatable_stop() -> No assert routes._cleanup_task_scope.tasks() == () +@pytest.mark.parametrize("finalizer_fails", [False, True]) +async def test_webrtc_cleanup_reports_only_unexpected_results_with_task_name( + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + *, + finalizer_fails: bool, +) -> None: + routes = WebRTCRoutes( + WebRTCTransportConfig(static_dir=None), + auth=None, + config_factory=lambda _transport: _GracefulSession(), # type: ignore[arg-type] + gate=CapacityGate(max_sessions=1), + manager=SessionManager(), + runtime_feedback=False, + ) + + async def finalize(_key: int, *, force: bool) -> None: + if force and finalizer_fails: + raise RuntimeError("forced finalizer failed") + + monkeypatch.setattr(routes, "_finalize_session_cleanup", finalize) + cleanup = routes._start_cleanup_task( + 7, + _NeverClosedTransport(), # type: ignore[arg-type] + ) + + with caplog.at_level("ERROR", logger="easycat.server.webrtc_routes"): + await routes.cancel_cleanup_tasks() + + messages = [record.getMessage() for record in caplog.records] + expected = "WebRTC cleanup task easycat-webrtc-force-cleanup-7 failed" + if finalizer_fails: + assert expected in messages + else: + assert not any(message.startswith("WebRTC cleanup task ") for message in messages) + assert cleanup.cancelled() + assert routes._cleanup_task_scope.tasks() == () + assert routes._force_cleanup_task_scope.tasks() == () + + async def test_drain_with_no_active_sessions_is_a_noop() -> None: gate: CapacityGate[int] = CapacityGate(max_sessions=4) await gate.drain(list, drain_timeout_s=1.0, force_after=True) From 15e6fbd4d26a6a4bfc0fad08ab6400bc7d2b0879 Mon Sep 17 00:00:00 2001 From: Yi Ding Date: Thu, 6 Aug 2026 20:07:10 -0700 Subject: [PATCH 2/2] webrtc: report late cleanup failures --- src/easycat/server/webrtc_routes.py | 17 +++++++++ tests/server/test_capacity_gate_drain.py | 44 ++++++++++++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/src/easycat/server/webrtc_routes.py b/src/easycat/server/webrtc_routes.py index cf3ff1712..bea749d30 100644 --- a/src/easycat/server/webrtc_routes.py +++ b/src/easycat/server/webrtc_routes.py @@ -189,6 +189,7 @@ def __init__( logger=logger, failure_message="WebRTC offer cleanup task failed", drop_if_closed=False, + release_standalone_when_idle=True, ) self._force_cleanup_task_scope = RuntimeTaskScope( owner_label="webrtc-routes-force-cleanup", @@ -197,6 +198,7 @@ def __init__( logger=logger, failure_message="WebRTC forced cleanup task failed", drop_if_closed=False, + release_standalone_when_idle=True, ) self._cleanup_task_keys: dict[asyncio.Task[Any], int] = {} self._released_cleanup_keys: set[int] = set() @@ -598,6 +600,9 @@ async def cancel_cleanup_tasks(self, *, timeout_s: float | None = None) -> None: results, explicitly_cancelled={task for task, _key in pending}, ) + else: + for finalizer in finalizers: + finalizer.add_done_callback(self._report_late_cleanup_result) await self._cleanup_task_scope.release_standalone_if_empty() await self._force_cleanup_task_scope.release_standalone_if_empty() @@ -620,6 +625,18 @@ def _report_cleanup_results( exc_info=result, ) + def _report_late_cleanup_result(self, task: asyncio.Task[Any]) -> None: + """Report a forced finalizer that settles after the hard deadline.""" + if task.cancelled(): + return + error = task.exception() + if error is not None: + logger.error( + "WebRTC cleanup task %s failed", + task.get_name(), + exc_info=error, + ) + async def _stop_managed_session(self, key: int, force: bool) -> None: """Route drain teardown through the manager's keyed stop ownership.""" await self._manager.remove(key, force=force) diff --git a/tests/server/test_capacity_gate_drain.py b/tests/server/test_capacity_gate_drain.py index c055af012..01219685d 100644 --- a/tests/server/test_capacity_gate_drain.py +++ b/tests/server/test_capacity_gate_drain.py @@ -777,6 +777,50 @@ async def finalize(_key: int, *, force: bool) -> None: assert routes._force_cleanup_task_scope.tasks() == () +async def test_webrtc_cleanup_reports_forced_finalizer_failure_after_timeout( + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + routes = WebRTCRoutes( + WebRTCTransportConfig(static_dir=None), + auth=None, + config_factory=lambda _transport: _GracefulSession(), # type: ignore[arg-type] + gate=CapacityGate(max_sessions=1), + manager=SessionManager(), + runtime_feedback=False, + ) + cancellation_seen = asyncio.Event() + release = asyncio.Event() + + async def finalize(_key: int, *, force: bool) -> None: + assert force is True + try: + await release.wait() + except asyncio.CancelledError: + cancellation_seen.set() + await release.wait() + raise RuntimeError("late forced finalizer failed") + + monkeypatch.setattr(routes, "_finalize_session_cleanup", finalize) + routes._start_cleanup_task( + 7, + _NeverClosedTransport(), # type: ignore[arg-type] + ) + + with caplog.at_level("ERROR", logger="easycat.server.webrtc_routes"): + await routes.cancel_cleanup_tasks(timeout_s=0.0) + await cancellation_seen.wait() + release.set() + for _ in range(20): + if routes._force_cleanup_task_scope.tasks() == (): + break + await asyncio.sleep(0) + + assert "WebRTC cleanup task easycat-webrtc-force-cleanup-7 failed" in caplog.text + assert "late forced finalizer failed" in caplog.text + assert routes._force_cleanup_task_scope.tasks() == () + + async def test_drain_with_no_active_sessions_is_a_noop() -> None: gate: CapacityGate[int] = CapacityGate(max_sessions=4) await gate.drain(list, drain_timeout_s=1.0, force_after=True)