chore: drain active calls before rolling updates (#474)

* chore: drain active calls before rolling updates

* fix: add a devops secret header

* fix: implement PR review

api/services/pipecat/active_calls.py

@@ -0,0 +1,35 @@
+"""In-process registry of active pipeline runs (live voice calls).
+
+Each uvicorn worker tracks the calls it is currently running so a deploy
+orchestrator can *drain* the worker before stopping it: poll the count, wait for
+zero, then send SIGTERM. Sending SIGTERM while calls are live makes uvicorn
+force-close their WebSockets (close code 1012), which cuts the calls instead of
+letting them finish — so the wait has to happen first.
+
+The registry is deliberately per-process. That is exactly the unit that gets
+drained: one uvicorn process per VM port (see ``scripts/rolling_update.sh``) or
+one uvicorn process per Kubernetes pod (drained via a ``preStop`` hook). The
+count is exposed read-only at ``GET /api/v1/health/active-calls`` and is also a
+natural autoscaling signal (concurrent calls per worker).
+
+Access is single-threaded (asyncio event loop), so no lock is needed. A set of
+run ids — rather than a bare counter — keeps register/unregister idempotent and
+makes the in-flight runs inspectable for debugging.
+"""
+
+_active_run_ids: set[int] = set()
+
+
+def register_active_call(workflow_run_id: int) -> None:
+    """Mark a pipeline run as active in this worker."""
+    _active_run_ids.add(workflow_run_id)
+
+
+def unregister_active_call(workflow_run_id: int) -> None:
+    """Mark a pipeline run as finished in this worker."""
+    _active_run_ids.discard(workflow_run_id)
+
+
+def active_call_count() -> int:
+    """Number of pipeline runs currently active in this worker."""
+    return len(_active_run_ids)

api/services/pipecat/run_pipeline.py

@@ -11,6 +11,10 @@ from api.services.integrations import (
     IntegrationRuntimeContext,
     create_runtime_sessions,
 )
+from api.services.pipecat.active_calls import (
+    register_active_call,
+    unregister_active_call,
+)
 from api.services.pipecat.audio_config import AudioConfig, create_audio_config
 from api.services.pipecat.event_handlers import (
     register_audio_data_handler,
@@ -163,6 +167,34 @@ async def run_pipeline_telephony(
     user_id: int,
     call_id: str,
     transport_kwargs: dict,
+) -> None:
+    """Run a pipeline for any telephony provider."""
+    # Register before any async setup so deploy drains see calls that are still
+    # resolving DB/config/transport state.
+    register_active_call(workflow_run_id)
+    try:
+        await _run_pipeline_telephony_impl(
+            websocket,
+            provider_name=provider_name,
+            workflow_id=workflow_id,
+            workflow_run_id=workflow_run_id,
+            user_id=user_id,
+            call_id=call_id,
+            transport_kwargs=transport_kwargs,
+        )
+    finally:
+        unregister_active_call(workflow_run_id)
+
+
+async def _run_pipeline_telephony_impl(
+    websocket,
+    *,
+    provider_name: str,
+    workflow_id: int,
+    workflow_run_id: int,
+    user_id: int,
+    call_id: str,
+    transport_kwargs: dict,
 ) -> None:
     """Run a pipeline for any telephony provider.
 
@@ -236,7 +268,7 @@ async def run_pipeline_telephony(
     )
 
     try:
-        await _run_pipeline(
+        await _run_pipeline_impl(
             transport,
             workflow_id,
             workflow_run_id,
@@ -260,6 +292,31 @@ async def run_pipeline_smallwebrtc(
     user_id: int,
     call_context_vars: dict = {},
     user_provider_id: str | None = None,
+) -> None:
+    """Run pipeline for WebRTC connections."""
+    # Register before any async setup so deploy drains see calls that are still
+    # resolving DB/config/transport state.
+    register_active_call(workflow_run_id)
+    try:
+        await _run_pipeline_smallwebrtc_impl(
+            webrtc_connection,
+            workflow_id,
+            workflow_run_id,
+            user_id,
+            call_context_vars=call_context_vars,
+            user_provider_id=user_provider_id,
+        )
+    finally:
+        unregister_active_call(workflow_run_id)
+
+
+async def _run_pipeline_smallwebrtc_impl(
+    webrtc_connection: SmallWebRTCConnection,
+    workflow_id: int,
+    workflow_run_id: int,
+    user_id: int,
+    call_context_vars: dict = {},
+    user_provider_id: str | None = None,
 ) -> None:
     """Run pipeline for WebRTC connections"""
     logger.debug(
@@ -309,7 +366,7 @@ async def run_pipeline_smallwebrtc(
         ambient_noise_config,
         is_realtime=is_realtime,
     )
-    await _run_pipeline(
+    await _run_pipeline_impl(
         transport,
         workflow_id,
         workflow_run_id,
@@ -332,6 +389,35 @@ async def _run_pipeline(
     user_provider_id: str | None = None,
     workflow_run=None,
     resolved_user_config=None,
+) -> None:
+    """Run the pipeline with active-call drain accounting."""
+    register_active_call(workflow_run_id)
+    try:
+        await _run_pipeline_impl(
+            transport,
+            workflow_id,
+            workflow_run_id,
+            user_id,
+            call_context_vars=call_context_vars,
+            audio_config=audio_config,
+            user_provider_id=user_provider_id,
+            workflow_run=workflow_run,
+            resolved_user_config=resolved_user_config,
+        )
+    finally:
+        unregister_active_call(workflow_run_id)
+
+
+async def _run_pipeline_impl(
+    transport,
+    workflow_id: int,
+    workflow_run_id: int,
+    user_id: int,
+    call_context_vars: dict = {},
+    audio_config: AudioConfig = None,
+    user_provider_id: str | None = None,
+    workflow_run=None,
+    resolved_user_config=None,
 ) -> None:
     """
     Run the pipeline with the given transport and configuration