From f9eb447d82665db00ad9568e551dbec68d623649 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 24 Sep 2025 08:24:28 -0700 Subject: [PATCH] FrameProcessor: timeout when cancelling tasks --- CHANGELOG.md | 6 ++++++ src/pipecat/processors/frame_processor.py | 6 +++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fc482e1b..5d75f9200 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -106,6 +106,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue where the pipeline could freeze if a task cancellation never + completed because a third-party library swallowed asyncio.CancelledError. We + now apply a timeout to task cancellations to prevent these freezes. If the + timeout is reached, the system logs warnings and leaves dangling tasks behind, + which can help diagnose where cancellation is being blocked. + - Fixed an `AudioBufferProcessor` issues that was causing user audio to be missing in stereo recordings causing bot and user overlaps. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 894dabbe7..0b55082aa 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -455,9 +455,13 @@ class FrameProcessor(BaseObject): name = f"{self}::{coroutine.cr_code.co_name}" return self.task_manager.create_task(coroutine, name) - async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None): + async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = 1.0): """Cancel a task managed by this processor. + A default timeout if 1 second is used in order to avoid potential + freezes caused by certain libraries that swallow + `asyncio.CancelledError`. + Args: task: The task to cancel. timeout: Optional timeout for task cancellation.