fix: bound PipelineWorker._wait_for_pipeline_end for EndFrame/StopFrame - #5211
fix: bound PipelineWorker._wait_for_pipeline_end for EndFrame/StopFrame#5211johnparkdd wants to merge 2 commits into
Conversation
PipelineWorker._wait_for_pipeline_end bounded only the CancelFrame wait with cancel_timeout_secs; the EndFrame/StopFrame branch waited on _pipeline_end_event indefinitely. If a processor never forwarded the finalize frame (e.g. a transport torn down mid-shutdown), the push-queue loop wedged forever and every later CancelFrame (end-call watchdog, idle auto-cancel, hard-timeout force-cancel) dead-lettered unread. Give the non-CancelFrame branch the same cancel_timeout_secs bound, mirroring the existing CancelFrame warning on timeout, then fall through to the existing clear/set tail. Stock semantics preserved: on_pipeline_finished is still called only at the sink for EndFrame/StopFrame (not on the timed-out path). Co-authored-by: oh-my-pi <https://omp.sh>
Co-authored-by: oh-my-pi <https://omp.sh>
|
Additional verification — full production end-of-call frame sequence, not just the direct Harness drives a real
Assertions: 6/6 PASS on unpatched (sequence reproduces the zombie), 8/8 PASS on this branch (sequence self-resolves, no rescue, clean teardown). Scenario |
Codecov Report✅ All modified and coverable lines are covered by tests.
... and 3 files with indirect coverage changes 🚀 New features to boost your workflow:
|
1 participant
Problem
PipelineWorker._process_push_queueawaits_wait_for_pipeline_end(frame)after pushing a finalize frame, and_pipeline_end_eventis set only when the frame reaches the worker sink (_sink_push_frame). If a processor never forwards the terminal frame downstream — e.g. its transport/websocket is torn down in the same second an end-of-call sequence begins — theEndFrame/StopFramebranch waits forever: onlyCancelFramehad a bounded wait (cancel_timeout_secs). The push-queue reader wedges inside thatawait, and every subsequentworker.cancel()enqueues aCancelFramethat is never read (dead letter). The pipeline zombies until an external task cancellation.We hit this in production on DoorDash's voice pipelines: an end-of-call races RTVI/transport teardown, the
EndFrameis lost, and the 30-second graceful-shutdown watchdog's cancel dead-letters — pipelines zombie for 16–20 minutes until an outer hard timeout kills them (dozens/day, spiking to 15–20/hour during an incident window).Fix
Give the
EndFrame/StopFramebranch the same bounded waitCancelFramealready has (cancel_timeout_secs, default 20s), with the same timeout warning. Success-path behavior is unchanged; on timeout the worker proceeds to the normal cleanup tail (clear()+_finished_event.set()) instead of hanging forever. This also un-dead-letters the fallback: once the bounded wait expires, the queue loop exits and cleanup runs.Repro / verification
Standalone harness driving a real
PipelineWorkerwith a processor that swallows theEndFrame(never forwards it — the production failure class):EndFrameswallowed mid-pipelinecancel_timeout(zombie)cancel_timeoutworker.cancel()Tests
test_task_end_frame_swallowed_finishes_after_cancel_timeout— swallowedEndFrame; worker finishes ≈cancel_timeoutwith the timeout warning, instead of hanging.test_task_stop_frame_swallowed_finishes_after_cancel_timeout— same forStopFrame.Notes
cancel_timeout_secs(default 20s) — the intended improvement; the warning log names the frame and suspects a blocked processor.