From 8025ed8aa077e51756cc3195bb00ea320b93e610 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Mon, 7 Sep 2026 08:27:52 +0800 Subject: [PATCH] fix: harden realtime pipeline lifecycle --- backend/requirements.txt | 2 + backend/routes/voice_webrtc.py | 5 +- backend/services/config_resolver.py | 3 +- backend/services/conversation_history.py | 60 +++++++++++ backend/services/pipecat/pipeline.py | 41 +------- backend/services/pipecat/pipeline_runner.py | 56 ++++++++++ backend/services/pipecat/processors.py | 2 +- backend/services/realtime/lifecycle.py | 29 +++++- backend/tests/test_config_resolver.py | 38 +++++++ backend/tests/test_conversation_history.py | 85 ++++++++++++++++ backend/tests/test_pipeline_runtime.py | 101 +++++++++++++++++++ backend/tests/test_voice_webrtc_lifecycle.py | 65 ++++++++++++ 12 files changed, 445 insertions(+), 42 deletions(-) create mode 100644 backend/services/pipecat/pipeline_runner.py create mode 100644 backend/tests/test_config_resolver.py create mode 100644 backend/tests/test_pipeline_runtime.py diff --git a/backend/requirements.txt b/backend/requirements.txt index 9d44e51..680c73c 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -18,6 +18,8 @@ python-dotenv pydantic loguru websockets>=13 +# websockets uses the system proxy; SOCKS connections need this optional extra. +python-socks[asyncio]>=2.4,<3 # 存储:Postgres(SQLAlchemy 2.0 异步 + asyncpg 驱动) sqlalchemy[asyncio]>=2.0 diff --git a/backend/routes/voice_webrtc.py b/backend/routes/voice_webrtc.py index 0582c37..10f04fe 100644 --- a/backend/routes/voice_webrtc.py +++ b/backend/routes/voice_webrtc.py @@ -145,7 +145,10 @@ async def http_offer(request: Request, payload: dict = Body(...)): # Client capabilities have one canonical transport: the request header. "client_tool_names": client_tool_names, } - answer = await _handle_offer_payload(offer_payload, _http_peers) + try: + answer = await _handle_offer_payload(offer_payload, _http_peers) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc return answer diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index 43ebab2..c342629 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -34,7 +34,7 @@ async def _resource_for( binding = await session.get(AssistantModelBinding, (assistant_id, capability)) resource_id = binding.model_resource_id if binding else None resource = await session.get(ModelResource, resource_id) if resource_id else None - if resource and resource.capability != capability: + if resource and (not resource.enabled or resource.capability != capability): resource = None if resource is None: resource = ( @@ -60,6 +60,7 @@ async def _agent_resource_for( resource = await session.get(ModelResource, resource_id) if resource_id else None if ( resource + and resource.enabled and resource.capability == "Agent" and resource.interface_type == interface_type ): diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py index 02f5383..d48eadd 100644 --- a/backend/services/conversation_history.py +++ b/backend/services/conversation_history.py @@ -16,6 +16,8 @@ from services.object_storage import delete_object, put_object, storage_uri MAX_WORKFLOW_TRACE_EVENTS = 1000 +MAX_PENDING_HISTORY_MESSAGES = 256 +HISTORY_FLUSH_TIMEOUT_SECONDS = 3.0 def _parse_timestamp(value: object) -> datetime: @@ -39,6 +41,63 @@ class ConversationRecorder: self._lock = asyncio.Lock() self._seen_events: set[str] = set() self._pending_artifacts: set[asyncio.Task[None]] = set() + self._pending_messages: asyncio.Queue[dict] = asyncio.Queue( + maxsize=MAX_PENDING_HISTORY_MESSAGES, + ) + self._message_writer: asyncio.Task[None] | None = None + self._accepting_messages = True + + def record_transport_message_later(self, message: object) -> None: + """Keep database latency off the media path, with bounded buffering.""" + if ( + not self._accepting_messages + or not isinstance(message, dict) + or message.get("type") not in { + "transcript", "assistant-text-end", "workflow-event", + } + ): + return + try: + self._pending_messages.put_nowait(deepcopy(message)) + except asyncio.QueueFull: + logger.warning(f"对话历史写入积压，跳过本条记录: session={self.session_id}") + return + if self._message_writer is None: + self._message_writer = asyncio.create_task( + self._write_pending_messages(), + name=f"conversation-history-writer:{self.session_id}", + ) + + async def _write_pending_messages(self) -> None: + while True: + message = await self._pending_messages.get() + try: + await self.record_transport_message(message) + except Exception as exc: + logger.warning(f"对话历史写入失败: {type(exc).__name__}") + finally: + self._pending_messages.task_done() + + async def flush_pending_messages(self) -> None: + """Drain final transcripts before finishing the conversation row.""" + self._accepting_messages = False + writer = self._message_writer + if writer is None: + return + try: + await asyncio.wait_for( + self._pending_messages.join(), + timeout=HISTORY_FLUSH_TIMEOUT_SECONDS, + ) + except TimeoutError: + logger.warning(f"对话历史收尾超时，丢弃剩余缓冲: session={self.session_id}") + finally: + writer.cancel() + await asyncio.gather(writer, return_exceptions=True) + self._message_writer = None + while not self._pending_messages.empty(): + self._pending_messages.get_nowait() + self._pending_messages.task_done() @classmethod async def start( @@ -288,6 +347,7 @@ class ConversationRecorder: raise async def _finish(self, *, status: str) -> None: + await self.flush_pending_messages() pending_artifacts = list(self._pending_artifacts) if pending_artifacts: await asyncio.gather(*pending_artifacts, return_exceptions=True) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 645b7e1..49927af 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -103,7 +103,7 @@ from services.realtime.protocol import ( PipelineProtocolRuntime, RealtimeProviderControlFrame, ) -from pipecat.workers.runner import WorkerRunner +from services.pipecat.pipeline_runner import run_media_worker KNOWLEDGE_TOOL_NAME = "search_knowledge_base" @@ -260,6 +260,7 @@ async def run_pipeline( assistant_id=assistant_id, channel=channel, protocol_adapter=protocol_adapter, + client_tool_names=client_tool_names, ) return @@ -987,25 +988,7 @@ async def run_pipeline( submit_user_input=submit_user_input, client_tools=client_tools, ) - runner = WorkerRunner(handle_sigint=False) - run_status = "completed" - can_finalize_recorder = True - try: - await runner.add_workers(worker) - await runner.run() - except GeneratorExit: - # A coroutine being synchronously closed cannot perform async cleanup. - # The WebRTC route now prevents this by retaining every pipeline task; - # keep this guard so a future unmanaged caller cannot corrupt asyncpg. - run_status = "failed" - can_finalize_recorder = False - raise - except Exception: - run_status = "failed" - raise - finally: - if recorder and can_finalize_recorder: - await recorder.finish(status=run_status) + await run_media_worker(worker, recorder) logger.info("管线已结束") @@ -1018,6 +1001,7 @@ async def run_realtime_pipeline( assistant_id: str | None = None, channel: str = "webrtc", protocol_adapter: PipelineProtocolAdapter | None = None, + client_tool_names: set[str] | None = None, ) -> None: """Run a speech-to-speech model that owns ASR, reasoning, and synthesis.""" realtime = create_realtime_service( @@ -1239,20 +1223,5 @@ async def run_realtime_pipeline( vision_enabled=vision_enabled, vision_state=vision_state, ) - runner = WorkerRunner(handle_sigint=False) - run_status = "completed" - can_finalize_recorder = True - try: - await runner.add_workers(worker) - await runner.run() - except GeneratorExit: - run_status = "failed" - can_finalize_recorder = False - raise - except Exception: - run_status = "failed" - raise - finally: - if recorder and can_finalize_recorder: - await recorder.finish(status=run_status) + await run_media_worker(worker, recorder) logger.info("Realtime 管线已结束") diff --git a/backend/services/pipecat/pipeline_runner.py b/backend/services/pipecat/pipeline_runner.py new file mode 100644 index 0000000..794d7d5 --- /dev/null +++ b/backend/services/pipecat/pipeline_runner.py @@ -0,0 +1,56 @@ +"""Run a media worker and persist its actual terminal outcome.""" + +import asyncio + +from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame +from pipecat.pipeline.worker import PipelineWorker +from pipecat.workers.runner import WorkerRunner + +from services.conversation_history import ConversationRecorder + + +async def run_media_worker( + worker: PipelineWorker, + recorder: ConversationRecorder | None, +) -> None: + # A runner can return normally after a fatal ErrorFrame or a failed worker. + # Only an observed, expected terminal frame counts as normal completion. + status = "failed" + fatal_error = False + + @worker.event_handler("on_pipeline_error") + async def on_pipeline_error(_worker, frame: ErrorFrame): + nonlocal status, fatal_error + if frame.fatal: + fatal_error = True + status = "failed" + + @worker.event_handler("on_pipeline_finished") + async def on_pipeline_finished(_worker, frame): + nonlocal status + normal_end = isinstance(frame, EndFrame) or ( + isinstance(frame, CancelFrame) + and frame.reason in {"client_disconnected", "idle timeout"} + ) + if normal_end and not fatal_error: + status = "completed" + + runner = WorkerRunner(handle_sigint=False) + can_finalize = True + try: + await runner.add_workers(worker) + await runner.run() + # Pipecat's runner consumes cancellation while draining its workers. + # Preserve the owning server task's cancellation after that cleanup. + owner = asyncio.current_task() + if owner is not None and owner.cancelling(): + raise asyncio.CancelledError + except GeneratorExit: + can_finalize = False # Synchronously closed coroutines cannot await. + raise + except BaseException: + status = "failed" + raise + finally: + if recorder and can_finalize: + await recorder.finish(status=status) diff --git a/backend/services/pipecat/processors.py b/backend/services/pipecat/processors.py index 24bb138..cb9bdee 100644 --- a/backend/services/pipecat/processors.py +++ b/backend/services/pipecat/processors.py @@ -657,7 +657,7 @@ class ConversationHistoryProcessor(FrameProcessor): await super().process_frame(frame, direction) await self.push_frame(frame, direction) if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame): - await self._recorder.record_transport_message(frame.message) + self._recorder.record_transport_message_later(frame.message) class KnowledgeRetrievalProcessor(FrameProcessor): diff --git a/backend/services/realtime/lifecycle.py b/backend/services/realtime/lifecycle.py index 91c773d..1207a8f 100644 --- a/backend/services/realtime/lifecycle.py +++ b/backend/services/realtime/lifecycle.py @@ -12,13 +12,33 @@ from loguru import logger active_connections: set[object] = set() pipeline_tasks: set[asyncio.Task[None]] = set() connection_tasks: dict[object, asyncio.Task[None]] = {} +connection_cleanup_tasks: set[asyncio.Task[None]] = set() DEFAULT_CLOSE_GRACE_SECONDS = 10.0 +async def _disconnect_connection(connection: object) -> None: + """Release media even when construction failed before transport startup.""" + try: + async with asyncio.timeout(DEFAULT_CLOSE_GRACE_SECONDS): + await connection.disconnect() + except Exception as exc: + logger.warning(f"Realtime 连接清理失败: {type(exc).__name__}") + + def _consume_pipeline_result(task: asyncio.Task[None], connection: object) -> None: pipeline_tasks.discard(task) if connection_tasks.get(connection) is task: connection_tasks.pop(connection, None) + if connection in active_connections: + active_connections.discard(connection) + # Run after the pipeline task is done: closed handlers can safely wait + # for that task without making pipeline shutdown wait on itself. + cleanup = asyncio.create_task( + _disconnect_connection(connection), + name=f"realtime-disconnect:{getattr(connection, 'pc_id', id(connection))}", + ) + connection_cleanup_tasks.add(cleanup) + cleanup.add_done_callback(connection_cleanup_tasks.discard) try: error = task.exception() except asyncio.CancelledError: @@ -86,11 +106,15 @@ async def shutdown_active_sessions( ) -> None: connections = list(active_connections) if connections: + active_connections.difference_update(connections) await asyncio.gather( - *(connection.disconnect() for connection in connections), + *(_disconnect_connection(connection) for connection in connections), return_exceptions=True, ) - active_connections.difference_update(connections) + + cleanup_tasks = list(connection_cleanup_tasks) + if cleanup_tasks: + await asyncio.gather(*cleanup_tasks, return_exceptions=True) tasks = list(pipeline_tasks) if not tasks: @@ -106,4 +130,3 @@ async def shutdown_active_sessions( logger.info( f"Realtime 会话清理完成: normal={len(done)} cancelled={len(cancelled)}" ) - diff --git a/backend/tests/test_config_resolver.py b/backend/tests/test_config_resolver.py new file mode 100644 index 0000000..ee3bfda --- /dev/null +++ b/backend/tests/test_config_resolver.py @@ -0,0 +1,38 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +from services.config_resolver import _agent_resource_for, _resource_for + + +class DisabledResourceTest(unittest.IsolatedAsyncioTestCase): + async def test_disabled_binding_uses_enabled_fallback_for_models_and_agents(self): + for capability, interface_type in (("LLM", "openai-llm"), ("Agent", "dify")): + with self.subTest(capability=capability): + disabled = SimpleNamespace( + id="disabled", capability=capability, + interface_type=interface_type, enabled=False, + ) + fallback = SimpleNamespace(id="enabled", enabled=True) + result = Mock() + result.scalar_one_or_none.return_value = fallback + session = SimpleNamespace( + get=AsyncMock(side_effect=[SimpleNamespace(model_resource_id="disabled"), disabled]), + execute=AsyncMock(return_value=result), + ) + if capability == "Agent": + selected = await _agent_resource_for(session, "assistant", interface_type) + else: + selected = await _resource_for(session, "assistant", capability) + self.assertIs(selected, fallback) + statement = str(session.execute.await_args.args[0]) + self.assertIn("model_resources.enabled IS true", statement) + + async def test_enabled_binding_is_preserved(self): + resource = SimpleNamespace(id="enabled", capability="LLM", enabled=True) + session = SimpleNamespace( + get=AsyncMock(side_effect=[SimpleNamespace(model_resource_id=resource.id), resource]), + execute=AsyncMock(), + ) + self.assertIs(await _resource_for(session, "assistant", "LLM"), resource) + session.execute.assert_not_awaited() diff --git a/backend/tests/test_conversation_history.py b/backend/tests/test_conversation_history.py index 2b85974..ddca809 100644 --- a/backend/tests/test_conversation_history.py +++ b/backend/tests/test_conversation_history.py @@ -9,6 +9,91 @@ from services.conversation_history import ConversationRecorder class ConversationRecorderTest(unittest.IsolatedAsyncioTestCase): + async def test_background_messages_keep_order_and_ignore_streaming_deltas(self): + recorder = ConversationRecorder("conv_test") + recorder.record_transport_message = AsyncMock() + events = [ + {"type": "transcript", "content": "user"}, + {"type": "assistant-text-end", "content": "assistant"}, + ] + for event in events: + recorder.record_transport_message_later(event) + recorder.record_transport_message_later({"type": "assistant-text-delta"}) + await recorder.flush_pending_messages() + self.assertEqual( + [call.args[0] for call in recorder.record_transport_message.await_args_list], + events, + ) + self.assertIsNone(recorder._message_writer) + + async def test_slow_database_does_not_block_audio(self): + from pipecat.frames.frames import EndFrame, OutputAudioRawFrame, OutputTransportMessageUrgentFrame + from pipecat.pipeline.pipeline import Pipeline + from pipecat.pipeline.worker import PipelineWorker + from pipecat.processors.frame_processor import FrameProcessor + from pipecat.workers.runner import WorkerRunner + from services.pipecat.processors import ConversationHistoryProcessor + + entered, release, audio_received = asyncio.Event(), asyncio.Event(), asyncio.Event() + recorder = ConversationRecorder("conv_test") + + async def slow_write(message): + entered.set() + await release.wait() + + recorder.record_transport_message = slow_write + + class AudioSink(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, OutputAudioRawFrame): + audio_received.set() + await self.push_frame(frame, direction) + + worker = PipelineWorker( + Pipeline([ConversationHistoryProcessor(recorder), AudioSink()]), enable_rtvi=False, + ) + runner = WorkerRunner(handle_sigint=False) + await runner.add_workers(worker) + task = asyncio.create_task(runner.run()) + try: + await worker.queue_frame(OutputTransportMessageUrgentFrame(message={"type": "transcript"})) + await asyncio.wait_for(entered.wait(), 3) + await worker.queue_frame(OutputAudioRawFrame(audio=b"\0" * 320, sample_rate=16000, num_channels=1)) + await asyncio.wait_for(audio_received.wait(), 1) + self.assertFalse(release.is_set()) + finally: + release.set() + await recorder.flush_pending_messages() + await worker.queue_frame(EndFrame()) + await asyncio.wait_for(task, 3) + + async def test_history_buffer_is_bounded_and_slow_writer_is_cancelled(self): + started = asyncio.Event() + cancelled = asyncio.Event() + with patch("services.conversation_history.MAX_PENDING_HISTORY_MESSAGES", 2): + recorder = ConversationRecorder("conv_test") + + async def blocked_write(message): + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + recorder.record_transport_message = blocked_write + recorder.record_transport_message_later({"type": "transcript"}) + await started.wait() + for _ in range(4): + recorder.record_transport_message_later({"type": "transcript"}) + self.assertEqual(recorder._pending_messages.qsize(), 2) + with patch("services.conversation_history.HISTORY_FLUSH_TIMEOUT_SECONDS", 0.01): + await recorder.flush_pending_messages() + self.assertTrue(cancelled.is_set()) + self.assertTrue(recorder._pending_messages.empty()) + recorder.record_transport_message_later({"type": "transcript"}) + self.assertIsNone(recorder._message_writer) + async def test_completed_conversation_queues_enabled_analysis(self): conversation = SimpleNamespace( status="active", diff --git a/backend/tests/test_pipeline_runtime.py b/backend/tests/test_pipeline_runtime.py new file mode 100644 index 0000000..18b25d3 --- /dev/null +++ b/backend/tests/test_pipeline_runtime.py @@ -0,0 +1,101 @@ +"""Regression checks for whole-pipeline assembly and terminal outcomes.""" + +import asyncio +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from models import AssistantConfig +from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, StartFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineWorker +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from services.pipecat import pipeline +from services.pipecat.pipeline_runner import run_media_worker + + +class PipelineAssemblyTest(unittest.IsolatedAsyncioTestCase): + async def test_realtime_startup_preserves_client_capabilities(self): + brain = SimpleNamespace( + spec=SimpleNamespace(supported_runtime_modes={"realtime"}), + system_prompt=lambda cfg: "test", + greeting=AsyncMock(return_value=""), + setup_realtime=AsyncMock(), + on_session_update=AsyncMock(), + ) + transport = SimpleNamespace( + input=lambda: FrameProcessor(), + output=lambda: FrameProcessor(), + event_handler=lambda name: lambda handler: handler, + ) + with ( + patch.object(pipeline, "build_brain", return_value=brain), + patch.object(pipeline, "create_realtime_service", return_value=FrameProcessor()), + patch.object(pipeline.ConversationRecorder, "start", new=AsyncMock(return_value=None)), + patch.object(pipeline, "run_media_worker", new=AsyncMock()) as run, + ): + await pipeline.run_pipeline( + transport, AssistantConfig(runtimeMode="realtime"), + client_tool_names={"show_message"}, + ) + run.assert_awaited_once() + runtime = brain.setup_realtime.await_args.args[1] + self.assertTrue(runtime.client_tools.supports("show_message")) + self.assertFalse(runtime.client_tools.supports("unregistered_tool")) + + +class TerminalProcessor(FrameProcessor): + def __init__(self, terminal): + super().__init__() + self.terminal = terminal + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + if isinstance(frame, StartFrame): + if isinstance(self.terminal, ErrorFrame): + await self.push_frame(self.terminal, FrameDirection.UPSTREAM) + else: + await self.push_frame(self.terminal) + + +class PipelineOutcomeTest(unittest.IsolatedAsyncioTestCase): + async def test_fatal_error_is_failed_even_when_runner_returns_normally(self): + worker = PipelineWorker( + Pipeline([TerminalProcessor(ErrorFrame(error="provider failed", fatal=True))]), + enable_rtvi=False, + ) + recorder = SimpleNamespace(finish=AsyncMock()) + await asyncio.wait_for(run_media_worker(worker, recorder), 3) + recorder.finish.assert_awaited_once_with(status="failed") + + async def test_normal_end_and_client_disconnect_are_completed(self): + for terminal in (EndFrame(), CancelFrame(reason="client_disconnected")): + with self.subTest(terminal=type(terminal).__name__): + worker = PipelineWorker(Pipeline([Passthrough()]), enable_rtvi=False) + recorder = SimpleNamespace(finish=AsyncMock()) + await worker.queue_frame(terminal) + await asyncio.wait_for(run_media_worker(worker, recorder), 3) + recorder.finish.assert_awaited_once_with(status="completed") + + async def test_server_task_cancellation_remains_failed_and_propagates(self): + worker = PipelineWorker(Pipeline([Passthrough()]), enable_rtvi=False) + started = asyncio.Event() + + @worker.event_handler("on_pipeline_started") + async def on_started(_worker, _frame): + started.set() + + recorder = SimpleNamespace(finish=AsyncMock()) + task = asyncio.create_task(run_media_worker(worker, recorder)) + await asyncio.wait_for(started.wait(), 3) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await asyncio.wait_for(task, 3) + recorder.finish.assert_awaited_once_with(status="failed") + + +class Passthrough(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) diff --git a/backend/tests/test_voice_webrtc_lifecycle.py b/backend/tests/test_voice_webrtc_lifecycle.py index b1dc8e9..04269ef 100644 --- a/backend/tests/test_voice_webrtc_lifecycle.py +++ b/backend/tests/test_voice_webrtc_lifecycle.py @@ -8,12 +8,16 @@ from unittest.mock import AsyncMock, patch from fastapi import HTTPException from routes import voice_webrtc +from services.realtime import lifecycle from starlette.requests import Request class FakeConnection: pc_id = "pc_test" + def __init__(self): + self.disconnect = AsyncMock() + def encoded_header_json(value: object) -> str: payload = json.dumps(value).encode("utf-8") @@ -81,6 +85,16 @@ class WebRTCOfferMetadataTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(raised.exception.status_code, 400) + async def test_invalid_assistant_returns_actionable_bad_request(self): + with patch.object( + voice_webrtc, "_handle_offer_payload", + new=AsyncMock(side_effect=ValueError("助手不存在: missing")), + ): + with self.assertRaises(HTTPException) as raised: + await voice_webrtc.http_offer(offer_request(), {"type": "offer"}) + self.assertEqual(raised.exception.status_code, 400) + self.assertIn("助手不存在", raised.exception.detail) + class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase): async def asyncTearDown(self): @@ -89,6 +103,9 @@ class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase): task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) + await asyncio.sleep(0) + if lifecycle.connection_cleanup_tasks: + await asyncio.gather(*list(lifecycle.connection_cleanup_tasks)) voice_webrtc._pipeline_tasks.clear() voice_webrtc._connection_tasks.clear() voice_webrtc._active_connections.clear() @@ -111,6 +128,54 @@ class WebRTCPipelineLifecycleTest(unittest.IsolatedAsyncioTestCase): self.assertNotIn(task, voice_webrtc._pipeline_tasks) self.assertNotIn(connection, voice_webrtc._connection_tasks) + async def test_failed_startup_disconnects_and_releases_connection(self): + connection = FakeConnection() + + async def pipeline(): + raise ValueError("provider construction failed") + + task = voice_webrtc._start_pipeline_task(connection, pipeline()) + with self.assertRaises(ValueError): + await task + await asyncio.sleep(0) + if lifecycle.connection_cleanup_tasks: + await asyncio.gather(*list(lifecycle.connection_cleanup_tasks)) + connection.disconnect.assert_awaited_once() + self.assertNotIn(connection, voice_webrtc._active_connections) + self.assertNotIn(connection, voice_webrtc._connection_tasks) + + async def test_cleanup_closed_handler_can_wait_for_completed_pipeline(self): + connection = FakeConnection() + + async def pipeline(): + return None + + task = voice_webrtc._start_pipeline_task(connection, pipeline()) + + async def disconnect(): + await lifecycle.wait_for_pipeline_close( + task, connection_id=connection.pc_id, timeout=0.01, + ) + + connection.disconnect.side_effect = disconnect + await task + await asyncio.sleep(0) + if lifecycle.connection_cleanup_tasks: + await asyncio.wait_for( + asyncio.gather(*list(lifecycle.connection_cleanup_tasks)), 0.5, + ) + connection.disconnect.assert_awaited_once() + + async def test_already_closed_connection_is_not_disconnected_twice(self): + connection = FakeConnection() + release = asyncio.Event() + task = voice_webrtc._start_pipeline_task(connection, release.wait()) + voice_webrtc._active_connections.discard(connection) + release.set() + await task + await asyncio.sleep(0) + connection.disconnect.assert_not_awaited() + async def test_stuck_pipeline_is_cancelled_after_close_grace_period(self): connection = FakeConnection() cancelled = asyncio.Event()