From 86c6141580f26666f59c7674edc17d38b0f645b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Aug 2025 23:01:41 -0700 Subject: [PATCH] DailyTransport: handle future cancellation --- CHANGELOG.md | 3 +++ src/pipecat/transports/services/daily.py | 12 ++++++++---- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aa7f46d3e..f54ba1ab9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -81,6 +81,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `DailyTransport` issue that would result in an unhandled + `concurrent.futures.CancelledError` when a future is cancelled. + - Fixed a `RivaSTTService` issue that would result in an unhandled `concurrent.futures.CancelledError` when a future is cancelled when reading from the audio chunks from the incoming audio stream. diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 1ad443cb5..daf879be7 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -13,6 +13,7 @@ real-time communication features. import asyncio import time +from concurrent.futures import CancelledError as FuturesCancelledError from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from typing import Any, Awaitable, Callable, Dict, Mapping, Optional @@ -1320,10 +1321,13 @@ class DailyTransportClient(EventHandler): def _call_async_callback(self, queue: asyncio.Queue, callback, *args): """Queue a callback for async execution on the event loop.""" - future = asyncio.run_coroutine_threadsafe( - queue.put((callback, *args)), self._get_event_loop() - ) - future.result() + try: + future = asyncio.run_coroutine_threadsafe( + queue.put((callback, *args)), self._get_event_loop() + ) + future.result() + except FuturesCancelledError: + pass async def _callback_task_handler(self, queue: asyncio.Queue): """Handle queued callbacks from the specified queue."""