DeepgramFluxSTTService: a blocked websocket send (peer stops reading, socket in CLOSE_WAIT) stalls the whole pipeline with no error, no reconnect and no log line

Related to #5735 (same service, same ending: an open user turn that never closes and a call that cannot end), but the trigger is different, so filing it on its own: the TCP connection to Deepgram stopped being read, the STT service's send blocked without raising, and nothing in the service or the pipeline noticed.

pipecat 1.10.0, DeepgramFluxSTTService on a websocket transport at 8 kHz, websockets 17.1, one process per call. DeepgramFluxTTSService on the same process and the same host.

Observed

On a ten-minute test call the agent answered five turns normally and then went silent for the remaining seven and a half minutes. From the machine, during the silence:

• The bot process's Deepgram STT socket was in CLOSE_WAIT with 83,514 bytes in the kernel send queue (about five seconds of 8 kHz 16-bit audio), 11 retransmissions and the retransmission timeout backed off to 12 s. The peer had stopped acknowledging data.
• A second socket to the same host, carrying the TTS connection, was ESTABLISHED with an empty queue.
• The process was idle: main thread in ep_poll, every other thread in futex_wait, 1-2% of one core, load average 0.01. Nothing was spinning and nothing was CPU-throttled (we checked; the host's burst balance never fell below 79 seconds and the throttle counter stayed at 0).
• DeepgramFluxTTSService logged connection closed, but with an error: sent 1011 (internal error) keepalive ping timeout and reconnected on attempt 1, 35 seconds into the stall. DeepgramFluxSTTService logged nothing at all — no close, no error, no reconnect — from the "Connected to Flux" line at the start of the call to teardown. Our own on_connection_error handler never fired either.
• The pipeline was blocked, not just quiet: PipelineWorker logged heartbeat frame not received for more than 10.0 seconds every 10 s for the rest of the call (35 times), the idle handler's EndWorkerFrame fired eleven times without ending the call, and at teardown CancelFrame timed out reaching the end of the pipeline. The audio buffer processor, which sits near the end of the pipeline, also stopped: both channels of the recording are digital zero from the moment of the stall.

We do not subclass, wrap or patch the Flux STT service; it is constructed with settings and tag only, and the only thing we read off it is _should_interrupt for a log line. Our turn strategies are on the aggregator, not on the service.

What the source says

The two services differ in one constructor line:

• services/deepgram/flux/stt.py L254 — WebsocketService.__init__(self, reconnect_on_error=False)
• services/deepgram/flux/tts.py L120 — WebsocketService.__init__(self, **kwargs), so reconnect_on_error defaults to True

That flag decides what happens in services/websocket_service.py _maybe_try_reconnect L328-336: the TTS side reconnects from the receive loop, the STT side reports the error and breaks. The comment at stt.py L193-202 says the STT service relies on send_with_retry instead, and run_stt (L426) does call it.

send_with_retry (websocket_service.py L256-274) recovers a send that raises. A send that blocks is not covered. In websockets 17.1, Connection.send() goes through send_context(), which calls send_data() and then await self.drain() (websockets/asyncio/connection.py L914-915). drain() (L1062-1078) waits on a future that is resolved only by resume_writing() or connection_lost(), with no timeout. With the peer not acknowledging, the transport stays paused and that await never returns.

STTService.process_audio_frame calls await self.process_generator(self.run_stt(frame.audio)) inline (services/stt_service.py L464), so the blocked send holds the STT processor's own frame task. Every processor after it in the pipeline stops receiving frames, which matches the missed heartbeats, the un-aired idle line, the EndWorkerFrames and the silent recorder.

The same backpressure appears to disarm the library's liveness check: keepalive() (websockets/asyncio/connection.py L803-849) calls await self.ping(), which goes through the same send_context() and drain(). On a paused transport the keepalive task blocks in the ping send and never reaches its pong timeout, so the 1011 that the TTS connection produced could not be produced here.

Other paths we checked and ruled out for this shape: _watchdog_task_handler (flux/stt_base.py L465-491) sends silence through _transport_send_audio, which is the same blocking send; STTService._keepalive_task_handler (L748-774) only fires when no audio has been sent, and audio was being sent; _verify_connection (websocket_service.py L168-181) is only reached from a reconnect that never starts. reconnect_on_error is not reachable from the DeepgramFluxSTTService constructor — L254 sets it explicitly and forwards no **kwargs.

Proven and inferred

Proven: the socket state and queue depth, the idle process, the absence of CPU throttling, the TTS reconnecting while the STT logged nothing, the repeated heartbeat timeouts, the eleven EndWorkerFrames, the CancelFrame drain timeout, and the source lines above.

Inferred: that a drain() that never resolves inside run_stt is what held the pipeline. We did not capture a stack trace of the blocked task, so this is read from the state rather than observed directly. Also inferred: that the peer stopped acknowledging first and sent its FIN later — the capture showing CLOSE_WAIT was taken seven minutes into the stall.

Not known: why the Deepgram side stopped acknowledging. We are not asking pipecat to answer that.

Request

Is there a documented way for a Flux STT connection to survive a peer that stops reading? Concretely, either of these would cover it, and we would rather use a pipecat feature than carry anything of our own:

1. A bounded send on the STT path — asyncio.wait_for around the send in send_with_retry, or a write_limit/timeout the service sets when it connects — so a stalled write raises and the existing send_with_retry reconnect can do its job.
2. Acting on the transport's own signals rather than only on an exception from send: the CLOSE_WAIT state and the paused transport are both visible before anything raises.

A related question, since it may be simpler: is reconnect_on_error=False on the Flux STT service still needed now that _connect_websocket creates the receive task only once (stt.py L346-349), and should it be reachable from the constructor?

Happy to run whatever repro you want. We cannot reproduce this on demand — it needs the peer to stop reading — but we can simulate it against a local websocket server that accepts a connection and then stops reading, on a stock 1.10.0 example, if that is a useful shape.