From bf055843e6269a585f8aefcfddf8b8e9b3d23143 Mon Sep 17 00:00:00 2001 From: James Hush Date: Wed, 26 Nov 2025 10:11:19 +0100 Subject: [PATCH] Fix race condition in DeepgramFluxSTTService reconnection Moved _receive_task and _watchdog_task creation from _connect_websocket() to _connect() to prevent multiple coroutines from attempting to receive from the websocket simultaneously during reconnection. Previously, when reconnection occurred, _connect_websocket() would be called while the existing _receive_task was still running, causing both to try to receive from the websocket. This resulted in the error: 'cannot call recv while another coroutine is already running recv or recv_streaming'. Now tasks are created only once during initial connection, and reconnection only re-establishes the websocket connection itself. This matches the pattern used by other websocket services in the codebase. Fixes issue reported in 0.0.95 where reconnection attempts would fail with recv errors. --- src/pipecat/services/deepgram/flux/stt.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index a0d45afe4..4c458c7ab 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -183,6 +183,14 @@ class DeepgramFluxSTTService(WebsocketSTTService): """ await self._connect_websocket() + # Creating the receiver task (only created once during initial connection) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + # Creating the watchdog task (only created once during initial connection) + if not self._watchdog_task: + self._watchdog_task = self.create_task(self._watchdog_task_handler()) + async def _disconnect(self): """Disconnect from WebSocket and clean up tasks. @@ -235,16 +243,6 @@ class DeepgramFluxSTTService(WebsocketSTTService): additional_headers={"Authorization": f"Token {self._api_key}"}, ) - # Creating the receiver task - if not self._receive_task: - self._receive_task = self.create_task( - self._receive_task_handler(self._report_error) - ) - - # Creating the watchdog task - if not self._watchdog_task: - self._watchdog_task = self.create_task(self._watchdog_task_handler()) - # Now wait for the connection established event logger.debug("WebSocket connected, waiting for server confirmation...") await self._connection_established_event.wait()