diff --git a/docs/development/07-transports-and-production.md b/docs/development/07-transports-and-production.md index ffdb2250a..fbd742c14 100644 --- a/docs/development/07-transports-and-production.md +++ b/docs/development/07-transports-and-production.md @@ -186,6 +186,9 @@ task identity. HTTP site and runner cleanup stages use a third named cohort plus a stage-keyed retry ledger, preventing a later `stop()` from racing a second cleanup call against an operation that survived the prior hard deadline. +The final `SessionManager.stop_all(force=True)` sweep is named and scope-owned +as well, so a cancellation-resistant Session remains attached to the server's +cleanup owner after the hard deadline records an incomplete stop. Read [`server/transports.py`](../../src/easycat/server/transports.py) for `CapacityGate` and hard-timeout helpers, and diff --git a/src/easycat/server/voice_server.py b/src/easycat/server/voice_server.py index b86a93999..8168c03dc 100644 --- a/src/easycat/server/voice_server.py +++ b/src/easycat/server/voice_server.py @@ -81,6 +81,8 @@ _WS_HANDLER_COHORT = "voice-server-ws-handler" _LISTENER_CLEANUP_TASK = "voice_server_listener_cleanup" _LISTENER_CLEANUP_COHORT = "voice-server-listener-cleanup" +_SESSION_SWEEP_TASK = "voice_server_session_sweep" +_SESSION_SWEEP_COHORT = "voice-server-session-sweep" class VoiceServer: @@ -112,6 +114,14 @@ def __init__( # Bare session registry only: add/remove/stop_all/connection. Capacity # and draining are NOT attributed to it (it has neither). self._manager: SessionManager[int] = SessionManager() + self._session_sweep_task_scope = RuntimeTaskScope( + owner_label="voice-server-session-sweep", + member_name=_SESSION_SWEEP_TASK, + cohort=_SESSION_SWEEP_COHORT, + logger=logger, + failure_message="VoiceServer SessionManager sweep task failed", + drop_if_closed=False, + ) # Shared capacity + draining collaborator (the M5 lift). It owns the # reservation counter, the active-connection set, and the draining flag @@ -605,7 +615,11 @@ async def _stop_unlocked(self, *, force: bool = False) -> None: # retries it after the handler has unwound. Bound the sweep with # ``force_shutdown_timeout_s`` so a force-stop that never returns cannot # block server teardown. - sweep_task = asyncio.create_task(self._manager.stop_all(force=True)) + sweep_task = self._session_sweep_task_scope.create_task( + self._manager.stop_all(force=True), + task_name="easycat-voice-server-session-sweep", + ) + assert sweep_task is not None sweep_succeeded, swept = await self._attempt_cleanup( "SessionManager hard sweep", _await_with_hard_timeout( @@ -620,6 +634,7 @@ async def _stop_unlocked(self, *, force: bool = False) -> None: report=sweep_task.result() if swept else None, cleanup_errors=cleanup_errors, ) + await self._session_sweep_task_scope.release_standalone_if_empty() # Keep session/resource references when any cleanup stage failed so a # later stop can retry them. The gate itself is always reset to a diff --git a/tests/ratchets/source-baseline.json b/tests/ratchets/source-baseline.json index 9cfd982e3..bf8d6e918 100644 --- a/tests/ratchets/source-baseline.json +++ b/tests/ratchets/source-baseline.json @@ -1,12 +1,12 @@ { "version": 1, - "rationale": "VoiceServer cancellation-resistant listener cleanup now uses named RuntimeTaskScope ownership.", + "rationale": "VoiceServer final SessionManager hard sweeps now use named RuntimeTaskScope ownership.", "counts": { "cancelled_error_handler": 100, "epoch_field": 17, "gather_return_exceptions": 22, "module_task_set": 1, - "raw_task_spawn": 30, + "raw_task_spawn": 29, "shield_loop": 4, "task_cancelling": 51 }, @@ -165,7 +165,6 @@ "raw_task_spawn\tserver/transports.py\t_await_with_hard_timeout\tasyncio.ensure_future\t2197525e3944cdd4\t0", "raw_task_spawn\tserver/transports.py\t_safe_await\tasyncio.ensure_future\t2197525e3944cdd4\t0", "raw_task_spawn\tserver/transports.py\tclose_websocket_connections\tasyncio.ensure_future\td0b144f338429d6b\t0", - "raw_task_spawn\tserver/voice_server.py\tVoiceServer._stop_unlocked\tasyncio.create_task\t17a66ec385918dba\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", "raw_task_spawn\tsession/_session.py\tSession._finish_interrupted_start\tasyncio.create_task\tc3c0859252009636\t0", diff --git a/tests/server/test_voice_server_lifecycle.py b/tests/server/test_voice_server_lifecycle.py index 12c310b01..4f350547c 100644 --- a/tests/server/test_voice_server_lifecycle.py +++ b/tests/server/test_voice_server_lifecycle.py @@ -650,6 +650,7 @@ async def stop_site() -> None: async def test_failed_session_hard_sweep_blocks_restart_and_retains_retry_ownership( caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, ) -> None: class FailingSession(_FakeSession): def __init__(self) -> None: @@ -671,6 +672,16 @@ async def stop(self, *, force: bool = False) -> None: server._active_session_objs[key] = session assert server._gate.try_acquire() server._gate.track(key) + sweep_task_names: list[str] = [] + stop_all = server._manager.stop_all + + async def named_stop_all(*, force: bool = False) -> object: + current = asyncio.current_task() + assert current is not None + sweep_task_names.append(current.get_name()) + return await stop_all(force=force) + + monkeypatch.setattr(server._manager, "stop_all", named_stop_all) with caplog.at_level(logging.ERROR), pytest.raises(RuntimeError, match="retained 1 session"): await server.stop(force=True) @@ -679,6 +690,8 @@ async def stop(self, *, force: bool = False) -> None: assert "session teardown failed" in caplog.text assert server._manager.get(key) is session assert server._active_session_objs == {key: session} + assert sweep_task_names == ["easycat-voice-server-session-sweep"] + assert server._session_sweep_task_scope.tasks() == () assert isinstance(server._lifecycle_cleanup_error, RuntimeError) with pytest.raises(RuntimeError, match="previous teardown cleanup is incomplete"): await server.start() @@ -692,6 +705,47 @@ async def stop(self, *, force: bool = False) -> None: assert server._lifecycle_cleanup_error is None +async def test_hard_sweep_timeout_keeps_named_task_owned_until_settlement( + monkeypatch: pytest.MonkeyPatch, +) -> None: + server = _idle_server( + enable_websocket=False, + enable_webrtc=False, + force_shutdown_timeout_s=0.01, + ) + server._started = True + sweep_started = asyncio.Event() + cancellation_seen = asyncio.Event() + release_sweep = asyncio.Event() + + async def stop_all(*, force: bool = False) -> object: + assert force is True + sweep_started.set() + while not release_sweep.is_set(): + try: + await release_sweep.wait() + except asyncio.CancelledError: + cancellation_seen.set() + return object() + + monkeypatch.setattr(server._manager, "stop_all", stop_all) + + with pytest.raises(RuntimeError, match="SessionManager.stop_all did not finish"): + await asyncio.wait_for(server.stop(force=True), timeout=0.5) + + assert sweep_started.is_set() + assert cancellation_seen.is_set() + tasks = server._session_sweep_task_scope.tasks() + assert len(tasks) == 1 + sweep_task = tasks[0] + assert sweep_task.get_name() == "easycat-voice-server-session-sweep" + + release_sweep.set() + await asyncio.wait_for(sweep_task, timeout=0.5) + await server._session_sweep_task_scope.release_standalone_if_empty() + assert server._session_sweep_task_scope.tasks() == () + + async def test_cancelled_stop_publishes_retryable_stopped_state_before_reraising() -> None: server = _idle_server(enable_websocket=False, enable_webrtc=False) site_stop_started = asyncio.Event()