diff --git a/api/constants.py b/api/constants.py index b2b5d24e..2712e488 100644 --- a/api/constants.py +++ b/api/constants.py @@ -150,7 +150,7 @@ COUNTRY_CODES = { "IE": "353", # Ireland } -DEFAULT_ORG_CONCURRENCY_LIMIT = os.getenv("DEFAULT_ORG_CONCURRENCY_LIMIT", 2) +DEFAULT_ORG_CONCURRENCY_LIMIT = int(os.getenv("DEFAULT_ORG_CONCURRENCY_LIMIT", "10")) DEFAULT_CAMPAIGN_RETRY_CONFIG = { "enabled": True, "max_retries": 1, diff --git a/api/errors/telephony_errors.py b/api/errors/telephony_errors.py index b7d18e01..a485bb9c 100644 --- a/api/errors/telephony_errors.py +++ b/api/errors/telephony_errors.py @@ -14,6 +14,7 @@ class TelephonyError(Enum): ACCOUNT_VALIDATION_FAILED = "ACCOUNT_VALIDATION_FAILED" PHONE_NUMBER_NOT_CONFIGURED = "PHONE_NUMBER_NOT_CONFIGURED" SIGNATURE_VALIDATION_FAILED = "SIGNATURE_VALIDATION_FAILED" + CONCURRENT_CALL_LIMIT = "CONCURRENT_CALL_LIMIT" QUOTA_EXCEEDED = "QUOTA_EXCEEDED" GENERAL_AUTH_FAILED = "GENERAL_AUTH_FAILED" VALID = "VALID" @@ -26,6 +27,7 @@ TELEPHONY_ERROR_MESSAGES = { TelephonyError.ACCOUNT_VALIDATION_FAILED: "Authentication error: Account credentials do not match. Please verify your account SID configuration in the dashboard matches your telephony provider settings.", TelephonyError.PHONE_NUMBER_NOT_CONFIGURED: "Phone number not configured: This number is not set up for inbound calls in your account. Please add this number to your telephony configuration.", TelephonyError.SIGNATURE_VALIDATION_FAILED: "Security error: Webhook signature validation failed. Please verify your auth token configuration and ensure requests are coming from your telephony provider.", + TelephonyError.CONCURRENT_CALL_LIMIT: "Service temporarily unavailable: Your account has reached its concurrent call limit. Please try again later.", TelephonyError.QUOTA_EXCEEDED: "Service temporarily unavailable: Your account has exceeded usage limits. Please contact your administrator or upgrade your plan to continue receiving calls.", TelephonyError.GENERAL_AUTH_FAILED: "Authentication failed: Please check your webhook URL configuration and ensure your telephony provider settings match your dashboard configuration.", } diff --git a/api/routes/agent_stream.py b/api/routes/agent_stream.py index 847c3228..105c4670 100644 --- a/api/routes/agent_stream.py +++ b/api/routes/agent_stream.py @@ -18,6 +18,10 @@ from starlette.websockets import WebSocketDisconnect from api.db import db_client from api.enums import CallType, WorkflowRunState +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + call_concurrency, +) from api.services.quota_service import authorize_workflow_run_start from api.services.telephony import registry as telephony_registry @@ -51,6 +55,16 @@ async def agent_stream_websocket( await websocket.close(code=1008, reason="Workflow not found") return + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + workflow.organization_id, + source=f"agent_stream:{provider_name}", + timeout=0, + ) + except CallConcurrencyLimitError: + await websocket.close(code=1008, reason="Concurrent call limit reached") + return + numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 workflow_run_name = f"WR-AGS-{numeric_suffix:08d}" initial_context = { @@ -58,14 +72,20 @@ async def agent_stream_websocket( "provider": provider_name, "direction": "inbound", } - workflow_run = await db_client.create_workflow_run( - workflow_run_name, - workflow.id, - provider_name, - user_id=workflow.user_id, - call_type=CallType.INBOUND, - initial_context=initial_context, - ) + try: + workflow_run = await db_client.create_workflow_run( + workflow_run_name, + workflow.id, + provider_name, + user_id=workflow.user_id, + call_type=CallType.INBOUND, + initial_context=initial_context, + organization_id=workflow.organization_id, + ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) + except Exception: + await call_concurrency.release_slot(concurrency_slot) + raise set_current_run_id(workflow_run.id) set_current_org_id(workflow.organization_id) @@ -79,36 +99,40 @@ async def agent_stream_websocket( f"agent-stream quota exceeded for user {workflow.user_id}: " f"{quota_result.error_message}" ) + await call_concurrency.release_workflow_run_slot(workflow_run.id) await websocket.close( code=1008, reason=quota_result.error_message or "Quota exceeded" ) return - await db_client.update_workflow_run( - run_id=workflow_run.id, state=WorkflowRunState.RUNNING.value - ) - - provider_instance = spec.provider_cls({}) try: - await provider_instance.handle_external_websocket( - websocket, - organization_id=workflow.organization_id, - workflow_id=workflow.id, - user_id=workflow.user_id, - workflow_run_id=workflow_run.id, - params=params, + await db_client.update_workflow_run( + run_id=workflow_run.id, state=WorkflowRunState.RUNNING.value ) - except NotImplementedError as e: - logger.warning(f"agent-stream provider {provider_name} not supported: {e}") + + provider_instance = spec.provider_cls({}) try: - await websocket.close(code=1011, reason=str(e)) - except RuntimeError: - pass - except WebSocketDisconnect as e: - logger.info(f"agent-stream disconnected: code={e.code} reason={e.reason}") - except Exception as e: - logger.error(f"agent-stream error for run {workflow_run.id}: {e}") - try: - await websocket.close(1011, "Internal server error") - except RuntimeError: - pass + await provider_instance.handle_external_websocket( + websocket, + organization_id=workflow.organization_id, + workflow_id=workflow.id, + user_id=workflow.user_id, + workflow_run_id=workflow_run.id, + params=params, + ) + except NotImplementedError as e: + logger.warning(f"agent-stream provider {provider_name} not supported: {e}") + try: + await websocket.close(code=1011, reason=str(e)) + except RuntimeError: + pass + except WebSocketDisconnect as e: + logger.info(f"agent-stream disconnected: code={e.code} reason={e.reason}") + except Exception as e: + logger.error(f"agent-stream error for run {workflow_run.id}: {e}") + try: + await websocket.close(1011, "Internal server error") + except RuntimeError: + pass + finally: + await call_concurrency.unregister_active_call(workflow_run.id) diff --git a/api/routes/public_agent.py b/api/routes/public_agent.py index 64706fb5..65222c56 100644 --- a/api/routes/public_agent.py +++ b/api/routes/public_agent.py @@ -14,6 +14,10 @@ from pydantic import BaseModel from api.db import db_client from api.enums import TriggerState, WorkflowStatus +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + call_concurrency, +) from api.services.quota_service import authorize_workflow_run_start from api.services.telephony.factory import ( get_default_telephony_provider, @@ -243,15 +247,32 @@ async def _execute_resolved_target( initial_context["api_key_created_by"] = api_key_created_by initial_context.update(request.initial_context or {}) - workflow_run = await db_client.create_workflow_run( - name=workflow_run_name, - workflow_id=target.workflow.id, - mode=workflow_run_mode, - initial_context=initial_context, - user_id=execution_user_id, - use_draft=use_draft, - organization_id=target.organization_id, - ) + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + target.organization_id, + source="public_agent", + timeout=0, + ) + except CallConcurrencyLimitError: + raise HTTPException( + status_code=429, + detail="Concurrent call limit reached", + ) + + try: + workflow_run = await db_client.create_workflow_run( + name=workflow_run_name, + workflow_id=target.workflow.id, + mode=workflow_run_mode, + initial_context=initial_context, + user_id=execution_user_id, + use_draft=use_draft, + organization_id=target.organization_id, + ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) + except Exception: + await call_concurrency.release_slot(concurrency_slot) + raise logger.info( f"Created workflow run {workflow_run.id} for public agent " @@ -267,10 +288,15 @@ async def _execute_resolved_target( workflow_run_id=workflow_run.id, ) if not quota_result.has_quota: + await call_concurrency.release_workflow_run_slot(workflow_run.id) raise HTTPException(status_code=402, detail=quota_result.error_message) # 9. Construct webhook URL for telephony provider callback - backend_endpoint, _ = await get_backend_endpoints() + try: + backend_endpoint, _ = await get_backend_endpoints() + except Exception: + await call_concurrency.release_workflow_run_slot(workflow_run.id) + raise webhook_endpoint = provider.WEBHOOK_ENDPOINT webhook_url = ( @@ -297,6 +323,7 @@ async def _execute_resolved_target( logger.warning( f"Failed to initiate call for workflow run {workflow_run.id}: {e}" ) + await call_concurrency.release_workflow_run_slot(workflow_run.id) raise HTTPException( status_code=400, detail=f"Failed to initiate call: {e}", diff --git a/api/routes/public_embed.py b/api/routes/public_embed.py index 500bf92e..616f9816 100644 --- a/api/routes/public_embed.py +++ b/api/routes/public_embed.py @@ -313,6 +313,7 @@ async def initialize_embed_session( **(init_request.context_variables or {}), "provider": WorkflowRunMode.SMALLWEBRTC.value, }, + organization_id=embed_token.organization_id, ) except Exception as e: logger.error(f"Failed to create workflow run: {e}") diff --git a/api/routes/telephony.py b/api/routes/telephony.py index 86cda43e..29b62359 100644 --- a/api/routes/telephony.py +++ b/api/routes/telephony.py @@ -25,6 +25,11 @@ from api.enums import CallType, WorkflowRunMode, WorkflowRunState from api.errors.telephony_errors import TelephonyError from api.sdk_expose import sdk_expose from api.services.auth.depends import get_user +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + WorkflowRunSlotAlreadyBoundError, + call_concurrency, +) from api.services.quota_service import authorize_workflow_run_start from api.services.telephony.call_transfer_manager import get_call_transfer_manager from api.services.telephony.factory import ( @@ -139,70 +144,6 @@ async def initiate_call( # Determine the workflow run mode based on provider type workflow_run_mode = provider.PROVIDER_NAME - workflow_run_id = request.workflow_run_id - - if not workflow_run_id: - # Merge template context variables (e.g. caller_number, called_number - # set in workflow settings for testing pre-call data fetch). - template_vars = workflow.template_context_variables or {} - - numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 - workflow_run_name = f"WR-TEL-OUT-{numeric_suffix:08d}" - workflow_run = await db_client.create_workflow_run( - workflow_run_name, - workflow.id, - workflow_run_mode, - user_id=execution_user_id, - call_type=CallType.OUTBOUND, - initial_context={ - **template_vars, - "phone_number": phone_number, - "called_number": phone_number, - "provider": provider.PROVIDER_NAME, - "telephony_configuration_id": telephony_configuration_id, - }, - use_draft=True, - organization_id=user.selected_organization_id, - ) - workflow_run_id = workflow_run.id - else: - workflow_run = await db_client.get_workflow_run( - workflow_run_id, organization_id=user.selected_organization_id - ) - if not workflow_run: - raise HTTPException(status_code=400, detail="Workflow run not found") - if workflow_run.workflow_id != workflow.id: - raise HTTPException( - status_code=400, - detail="workflow_run_workflow_mismatch", - ) - workflow_run_name = workflow_run.name - - # Check Dograh quota after the run exists so hosted v2 can mint and store - # the MPS correlation id before initiating the call. - quota_result = await authorize_workflow_run_start( - workflow_id=workflow.id, - workflow_run_id=workflow_run_id, - actor_user=user, - ) - if not quota_result.has_quota: - raise HTTPException(status_code=402, detail=quota_result.error_message) - - # Construct webhook URL based on provider type - backend_endpoint, _ = await get_backend_endpoints() - - webhook_endpoint = provider.WEBHOOK_ENDPOINT - - webhook_url = ( - f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}" - f"?workflow_id={workflow.id}" - f"&user_id={execution_user_id}" - f"&workflow_run_id={workflow_run_id}" - f"&organization_id={user.selected_organization_id}" - ) - - keywords = {"workflow_id": workflow.id, "user_id": execution_user_id} - # Resolve optional caller-ID. The config has already been validated against # the user's organization, so filtering by config_id is sufficient for # tenant isolation. @@ -220,14 +161,102 @@ async def initiate_call( raise HTTPException(status_code=400, detail="from_phone_number_not_found") from_number = phone_row.address_normalized - # Initiate call via provider - result = await provider.initiate_call( - to_number=phone_number, - webhook_url=webhook_url, + workflow_run_id = request.workflow_run_id + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + user.selected_organization_id, + source="telephony_outbound", + timeout=0, + ) + except CallConcurrencyLimitError: + raise HTTPException(status_code=429, detail="Concurrent call limit reached") + + try: + if not workflow_run_id: + # Merge template context variables (e.g. caller_number, called_number + # set in workflow settings for testing pre-call data fetch). + template_vars = workflow.template_context_variables or {} + + numeric_suffix = int(str(uuid.uuid4()).replace("-", "")[:8], 16) % 100000000 + workflow_run_name = f"WR-TEL-OUT-{numeric_suffix:08d}" + workflow_run = await db_client.create_workflow_run( + workflow_run_name, + workflow.id, + workflow_run_mode, + user_id=execution_user_id, + call_type=CallType.OUTBOUND, + initial_context={ + **template_vars, + "phone_number": phone_number, + "called_number": phone_number, + "provider": provider.PROVIDER_NAME, + "telephony_configuration_id": telephony_configuration_id, + }, + use_draft=True, + organization_id=user.selected_organization_id, + ) + workflow_run_id = workflow_run.id + else: + workflow_run = await db_client.get_workflow_run( + workflow_run_id, organization_id=user.selected_organization_id + ) + if not workflow_run: + raise HTTPException(status_code=400, detail="Workflow run not found") + if workflow_run.workflow_id != workflow.id: + raise HTTPException( + status_code=400, + detail="workflow_run_workflow_mismatch", + ) + workflow_run_name = workflow_run.name + + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run_id) + except WorkflowRunSlotAlreadyBoundError: + raise HTTPException( + status_code=409, + detail="Workflow run already has an active call", + ) + except Exception: + await call_concurrency.release_slot(concurrency_slot) + raise + + # Check Dograh quota after the run exists so hosted v2 can mint and store + # the MPS correlation id before initiating the call. + quota_result = await authorize_workflow_run_start( + workflow_id=workflow.id, workflow_run_id=workflow_run_id, - from_number=from_number, - **keywords, + actor_user=user, ) + if not quota_result.has_quota: + await call_concurrency.release_workflow_run_slot(workflow_run_id) + raise HTTPException(status_code=402, detail=quota_result.error_message) + + try: + # Construct webhook URL based on provider type + backend_endpoint, _ = await get_backend_endpoints() + + webhook_endpoint = provider.WEBHOOK_ENDPOINT + + webhook_url = ( + f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}" + f"?workflow_id={workflow.id}" + f"&user_id={execution_user_id}" + f"&workflow_run_id={workflow_run_id}" + f"&organization_id={user.selected_organization_id}" + ) + + keywords = {"workflow_id": workflow.id, "user_id": execution_user_id} + + # Initiate call via provider + result = await provider.initiate_call( + to_number=phone_number, + webhook_url=webhook_url, + workflow_run_id=workflow_run_id, + from_number=from_number, + **keywords, + ) + except Exception: + await call_concurrency.release_workflow_run_slot(workflow_run_id) + raise # Store provider metadata and caller_number in workflow run context gathered_context = { @@ -425,6 +454,7 @@ async def _create_inbound_workflow_run( normalized_data, telephony_configuration_id: int, from_phone_number_id: Optional[int] = None, + organization_id: int | None = None, ) -> int: """Create workflow run for inbound call and return run ID""" call_id = normalized_data.call_id @@ -456,6 +486,7 @@ async def _create_inbound_workflow_run( "raw_webhook_data": normalized_data.raw_data, }, }, + organization_id=organization_id, ) logger.info( @@ -655,6 +686,8 @@ async def _handle_telephony_websocket( except RuntimeError: # WebSocket already closed, ignore pass + finally: + await call_concurrency.unregister_active_call(workflow_run_id) @router.post("/inbound/run") @@ -765,40 +798,67 @@ async def handle_inbound_run(request: Request): TelephonyError.SIGNATURE_VALIDATION_FAILED ) - # 5. Create workflow run + authorize quota before returning provider - # stream instructions. - workflow_run_id = await _create_inbound_workflow_run( - workflow_id, - user_id, - provider_class.PROVIDER_NAME, - normalized_data, - telephony_configuration_id=telephony_configuration_id, - from_phone_number_id=phone_row.id, - ) - quota_result = await authorize_workflow_run_start( - workflow_id=workflow_id, - workflow_run_id=workflow_run_id, - ) - if not quota_result.has_quota: - logger.warning( - f"User {user_id} has exceeded quota: {quota_result.error_message}" + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + config.organization_id, + source=f"inbound:{provider_class.PROVIDER_NAME}", + timeout=0, ) + except CallConcurrencyLimitError: return provider_class.generate_validation_error_response( - TelephonyError.QUOTA_EXCEEDED + TelephonyError.CONCURRENT_CALL_LIMIT ) - backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() - websocket_url = ( - f"{wss_backend_endpoint}/api/v1/telephony/ws/" - f"{workflow_id}/{user_id}/{workflow_run_id}" - ) + workflow_run_id = None + try: + # 5. Create workflow run + authorize quota before returning provider + # stream instructions. + workflow_run_id = await _create_inbound_workflow_run( + workflow_id, + user_id, + provider_class.PROVIDER_NAME, + normalized_data, + telephony_configuration_id=telephony_configuration_id, + from_phone_number_id=phone_row.id, + organization_id=config.organization_id, + ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run_id) - return await provider_instance.start_inbound_stream( - websocket_url=websocket_url, - workflow_run_id=workflow_run_id, - normalized_data=normalized_data, - backend_endpoint=backend_endpoint, - ) + quota_result = await authorize_workflow_run_start( + workflow_id=workflow_id, + workflow_run_id=workflow_run_id, + ) + if not quota_result.has_quota: + logger.warning( + f"User {user_id} has exceeded quota: {quota_result.error_message}" + ) + await call_concurrency.release_workflow_run_slot(workflow_run_id) + return provider_class.generate_validation_error_response( + TelephonyError.QUOTA_EXCEEDED + ) + + backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() + websocket_url = ( + f"{wss_backend_endpoint}/api/v1/telephony/ws/" + f"{workflow_id}/{user_id}/{workflow_run_id}" + ) + + return await provider_instance.start_inbound_stream( + websocket_url=websocket_url, + workflow_run_id=workflow_run_id, + normalized_data=normalized_data, + backend_endpoint=backend_endpoint, + ) + except WorkflowRunSlotAlreadyBoundError: + return provider_class.generate_validation_error_response( + TelephonyError.CONCURRENT_CALL_LIMIT + ) + except Exception: + if workflow_run_id: + await call_concurrency.release_workflow_run_slot(workflow_run_id) + else: + await call_concurrency.release_slot(concurrency_slot) + raise except ValueError as e: logger.error(f"/inbound/run request parsing error: {e}") @@ -900,38 +960,72 @@ async def handle_inbound_telephony( logger.error(f"Request validation failed: {error_type}") return provider_class.generate_validation_error_response(error_type) - # Create workflow run. user_id = workflow_context["user_id"] - workflow_run_id = await _create_inbound_workflow_run( - workflow_id, - workflow_context["user_id"], - workflow_context["provider"], - normalized_data, - telephony_configuration_id=workflow_context["telephony_configuration_id"], - from_phone_number_id=workflow_context.get("from_phone_number_id"), - ) - quota_result = await authorize_workflow_run_start( - workflow_id=workflow_id, - workflow_run_id=workflow_run_id, - ) - if not quota_result.has_quota: - logger.warning( - f"User {user_id} has exceeded quota for inbound calls: {quota_result.error_message}" + organization_id = workflow_context["organization_id"] + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + organization_id, + source=f"inbound_legacy:{workflow_context['provider']}", + timeout=0, ) + except CallConcurrencyLimitError: return provider_class.generate_validation_error_response( - TelephonyError.QUOTA_EXCEEDED + TelephonyError.CONCURRENT_CALL_LIMIT ) - # Generate response URLs - backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() - websocket_url = f"{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{workflow_context['user_id']}/{workflow_run_id}" + workflow_run_id = None + try: + # Create workflow run. + workflow_run_id = await _create_inbound_workflow_run( + workflow_id, + workflow_context["user_id"], + workflow_context["provider"], + normalized_data, + telephony_configuration_id=workflow_context[ + "telephony_configuration_id" + ], + from_phone_number_id=workflow_context.get("from_phone_number_id"), + organization_id=organization_id, + ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run_id) - response = await provider_instance.start_inbound_stream( - websocket_url=websocket_url, - workflow_run_id=workflow_run_id, - normalized_data=normalized_data, - backend_endpoint=backend_endpoint, - ) + quota_result = await authorize_workflow_run_start( + workflow_id=workflow_id, + workflow_run_id=workflow_run_id, + ) + if not quota_result.has_quota: + logger.warning( + f"User {user_id} has exceeded quota for inbound calls: " + f"{quota_result.error_message}" + ) + await call_concurrency.release_workflow_run_slot(workflow_run_id) + return provider_class.generate_validation_error_response( + TelephonyError.QUOTA_EXCEEDED + ) + + # Generate response URLs + backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() + websocket_url = ( + f"{wss_backend_endpoint}/api/v1/telephony/ws/" + f"{workflow_id}/{workflow_context['user_id']}/{workflow_run_id}" + ) + + response = await provider_instance.start_inbound_stream( + websocket_url=websocket_url, + workflow_run_id=workflow_run_id, + normalized_data=normalized_data, + backend_endpoint=backend_endpoint, + ) + except WorkflowRunSlotAlreadyBoundError: + return provider_class.generate_validation_error_response( + TelephonyError.CONCURRENT_CALL_LIMIT + ) + except Exception: + if workflow_run_id: + await call_concurrency.release_workflow_run_slot(workflow_run_id) + else: + await call_concurrency.release_slot(concurrency_slot) + raise logger.info( f"Generated {normalized_data.provider} response for call {normalized_data.call_id}" diff --git a/api/routes/webrtc_signaling.py b/api/routes/webrtc_signaling.py index 75ed0482..6e376737 100644 --- a/api/routes/webrtc_signaling.py +++ b/api/routes/webrtc_signaling.py @@ -40,6 +40,11 @@ from api.routes.turn_credentials import ( generate_turn_credentials, ) from api.services.auth.depends import get_user_ws +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + WorkflowRunSlotAlreadyBoundError, + call_concurrency, +) from api.services.pipecat.run_pipeline import run_pipeline_smallwebrtc from api.services.pipecat.ws_sender_registry import ( register_ws_sender, @@ -321,6 +326,8 @@ class SignalingManager: workflow_id: int, workflow_run_id: int, user: UserModel, + enforce_call_concurrency: bool = False, + call_concurrency_source: str = "webrtc", ): """Handle WebSocket connection for signaling.""" await websocket.accept() @@ -338,6 +345,8 @@ class SignalingManager: workflow_run_id, user, connection_key, + enforce_call_concurrency, + call_concurrency_source, ) except WebSocketDisconnect: logger.info(f"WebSocket disconnected for {connection_id}") @@ -379,6 +388,8 @@ class SignalingManager: workflow_run_id: int, user: UserModel, connection_key: str, + enforce_call_concurrency: bool, + call_concurrency_source: str = "webrtc", ): """Handle incoming WebSocket messages.""" msg_type = message.get("type") @@ -386,7 +397,14 @@ class SignalingManager: if msg_type == "offer": await self._handle_offer( - ws, payload, workflow_id, workflow_run_id, user, connection_key + ws, + payload, + workflow_id, + workflow_run_id, + user, + connection_key, + enforce_call_concurrency, + call_concurrency_source, ) elif msg_type == "ice-candidate": await self._handle_ice_candidate(payload, connection_key) @@ -401,6 +419,8 @@ class SignalingManager: workflow_run_id: int, user: UserModel, connection_key: str, + enforce_call_concurrency: bool, + call_concurrency_source: str = "webrtc", ): """Handle offer message and create answer with ICE trickling.""" pc_id = payload.get("pc_id") @@ -472,69 +492,135 @@ class SignalingManager: } ) else: + concurrency_slot = None + concurrency_bound = False + pipeline_started = False + pc = None + if enforce_call_concurrency: + if org_id is None: + await ws.send_json( + { + "type": "error", + "payload": { + "error_type": "organization_not_found", + "message": "Workflow organization not found", + }, + } + ) + return + + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + org_id, + source=call_concurrency_source, + timeout=0, + ) + await call_concurrency.bind_workflow_run( + concurrency_slot, + workflow_run_id, + ) + concurrency_bound = True + except CallConcurrencyLimitError: + await ws.send_json( + { + "type": "error", + "payload": { + "error_type": "concurrency_limit_exceeded", + "message": "Concurrent call limit reached", + }, + } + ) + return + except WorkflowRunSlotAlreadyBoundError: + await ws.send_json( + { + "type": "error", + "payload": { + "error_type": "workflow_run_already_active", + "message": "Workflow run already has an active call", + }, + } + ) + return + # Create new connection using correct SmallWebRTC API # Generate ICE servers with time-limited TURN credentials for this user - user_ice_servers = get_ice_servers(user_id=str(user.id)) - pc = SmallWebRTCConnection( - ice_servers=user_ice_servers, connection_timeout_secs=60 - ) - # Set the pc_id before initialization so it's available in get_answer() - pc._pc_id = pc_id - - # Initialize connection with offer - await pc.initialize(sdp=sdp, type=type_) - - # Store peer connection using client's pc_id - self._track_peer_connection(connection_key, pc_id, pc) - - # Register WebSocket sender for real-time feedback - async def ws_sender(message: dict): - if ws.application_state == WebSocketState.CONNECTED: - await ws.send_json(message) - - register_ws_sender(workflow_run_id, ws_sender) - - # Setup closed handler - @pc.event_handler("closed") - async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): - logger.info(f"PeerConnection closed: {webrtc_connection.pc_id}") - owner_connection_id = self._forget_peer_connection( - webrtc_connection.pc_id + try: + user_ice_servers = get_ice_servers(user_id=str(user.id)) + pc = SmallWebRTCConnection( + ice_servers=user_ice_servers, connection_timeout_secs=60 ) - if owner_connection_id == connection_key: - await self._notify_call_ended_and_close_websocket( - ws, - workflow_run_id, - webrtc_connection.pc_id, - reason="peer_connection_closed", + # Set the pc_id before initialization so it's available in get_answer() + pc._pc_id = pc_id + + # Initialize connection with offer + await pc.initialize(sdp=sdp, type=type_) + + # Store peer connection using client's pc_id + self._track_peer_connection(connection_key, pc_id, pc) + + # Register WebSocket sender for real-time feedback + async def ws_sender(message: dict): + if ws.application_state == WebSocketState.CONNECTED: + await ws.send_json(message) + + register_ws_sender(workflow_run_id, ws_sender) + + # Setup closed handler + @pc.event_handler("closed") + async def handle_disconnected( + webrtc_connection: SmallWebRTCConnection, + ): + logger.info(f"PeerConnection closed: {webrtc_connection.pc_id}") + owner_connection_id = self._forget_peer_connection( + webrtc_connection.pc_id ) + if owner_connection_id == connection_key: + await self._notify_call_ended_and_close_websocket( + ws, + workflow_run_id, + webrtc_connection.pc_id, + reason="peer_connection_closed", + ) - # Start pipeline in background - asyncio.create_task( - run_pipeline_smallwebrtc( - pc, - workflow_id, - workflow_run_id, - user.id, - call_context_vars, - user_provider_id=str(user.provider_id), + # Start pipeline in background + asyncio.create_task( + run_pipeline_smallwebrtc( + pc, + workflow_id, + workflow_run_id, + user.id, + call_context_vars, + user_provider_id=str(user.provider_id), + ) ) - ) + pipeline_started = True - # Get answer after initialization - answer = pc.get_answer() + # Get answer after initialization + answer = pc.get_answer() - # Send answer immediately (ICE candidates will be sent separately via trickling) - await ws.send_json( - { - "type": "answer", - "payload": { - "sdp": filter_outbound_sdp(answer["sdp"]), - "type": answer["type"], - "pc_id": answer["pc_id"], - }, - } - ) + # Send answer immediately (ICE candidates will be sent separately via trickling) + await ws.send_json( + { + "type": "answer", + "payload": { + "sdp": filter_outbound_sdp(answer["sdp"]), + "type": answer["type"], + "pc_id": answer["pc_id"], + }, + } + ) + except Exception: + if pipeline_started and pc is not None: + try: + await pc.disconnect() + except Exception as e: + logger.debug(f"Failed to disconnect failed offer pc: {e}") + elif concurrency_bound: + await call_concurrency.release_workflow_run_slot(workflow_run_id) + elif concurrency_slot is not None: + await call_concurrency.release_slot(concurrency_slot) + raise async def _handle_ice_candidate(self, payload: dict, connection_key: str): """Handle incoming ICE candidate from client. @@ -636,7 +722,12 @@ async def signaling_websocket( raise HTTPException(status_code=400, detail="Bad workflow_run_id") await signaling_manager.handle_websocket( - websocket, workflow_id, workflow_run_id, user + websocket, + workflow_id, + workflow_run_id, + user, + enforce_call_concurrency=True, + call_concurrency_source="webrtc", ) @@ -693,5 +784,10 @@ async def public_signaling_websocket( # Handle the WebSocket connection using the existing signaling manager await signaling_manager.handle_websocket( - websocket, embed_token.workflow_id, embed_session.workflow_run_id, user + websocket, + embed_token.workflow_id, + embed_session.workflow_run_id, + user, + enforce_call_concurrency=True, + call_concurrency_source="public_embed", ) diff --git a/api/services/call_concurrency.py b/api/services/call_concurrency.py new file mode 100644 index 00000000..33bd526f --- /dev/null +++ b/api/services/call_concurrency.py @@ -0,0 +1,190 @@ +import asyncio +import time +from dataclasses import dataclass + +from loguru import logger + +from api.constants import DEFAULT_ORG_CONCURRENCY_LIMIT +from api.db import db_client +from api.enums import OrganizationConfigurationKey +from api.services.campaign.rate_limiter import rate_limiter + + +@dataclass(frozen=True) +class CallConcurrencySlot: + organization_id: int + slot_id: str + max_concurrent: int + source: str + + +class CallConcurrencyLimitError(Exception): + """Raised when an org has no available concurrent call slots.""" + + def __init__( + self, + *, + organization_id: int, + source: str, + wait_time: float, + max_concurrent: int, + ): + self.organization_id = organization_id + self.source = source + self.wait_time = wait_time + self.max_concurrent = max_concurrent + super().__init__( + f"Concurrent call limit reached for org {organization_id} " + f"(source={source}, limit={max_concurrent}, waited={wait_time:.1f}s)" + ) + + +class WorkflowRunSlotAlreadyBoundError(Exception): + """Raised when a workflow run already owns a concurrent call slot.""" + + def __init__(self, workflow_run_id: int): + self.workflow_run_id = workflow_run_id + super().__init__( + f"Workflow run {workflow_run_id} already has an active call slot" + ) + + +class CallConcurrencyService: + def __init__(self): + self.default_concurrent_limit = int(DEFAULT_ORG_CONCURRENCY_LIMIT) + + async def get_org_concurrent_limit(self, organization_id: int) -> int: + """Get the concurrent call limit for an organization.""" + try: + config = await db_client.get_configuration( + organization_id, + OrganizationConfigurationKey.CONCURRENT_CALL_LIMIT.value, + ) + if config and config.value: + value = config.value.get("value") + if value is not None: + return int(value) + except Exception as e: + logger.warning( + f"Error getting concurrent limit for org {organization_id}: {e}" + ) + return self.default_concurrent_limit + + async def acquire_org_slot( + self, + organization_id: int, + *, + source: str, + timeout: float = 0, + max_concurrent_override: int | None = None, + retry_interval: float = 1, + ) -> CallConcurrencySlot: + org_concurrent_limit = await self.get_org_concurrent_limit(organization_id) + if max_concurrent_override is None: + max_concurrent = org_concurrent_limit + else: + max_concurrent = min(int(max_concurrent_override), org_concurrent_limit) + + wait_start = time.time() + while True: + acquisition = await rate_limiter.try_acquire_concurrent_slot_details( + organization_id, max_concurrent + ) + if acquisition: + logger.info( + f"Acquired concurrent call slot for org {organization_id}: " + f"source={source}, active_calls=" + f"{acquisition.active_count}/{max_concurrent}, " + f"slot_id={acquisition.slot_id}" + ) + return CallConcurrencySlot( + organization_id=organization_id, + slot_id=acquisition.slot_id, + max_concurrent=max_concurrent, + source=source, + ) + + wait_time = time.time() - wait_start + if wait_time >= timeout: + current_count = await rate_limiter.get_concurrent_count(organization_id) + logger.warning( + f"Concurrent call limit reached for org {organization_id}: " + f"source={source}, active_calls={current_count}/{max_concurrent}, " + f"waited={wait_time:.1f}s" + ) + raise CallConcurrencyLimitError( + organization_id=organization_id, + source=source, + wait_time=wait_time, + max_concurrent=max_concurrent, + ) + + logger.debug( + f"Waiting for concurrent call slot for org {organization_id}, " + f"source={source}, waited {wait_time:.1f}s" + ) + await asyncio.sleep(min(retry_interval, max(0, timeout - wait_time))) + + async def bind_workflow_run( + self, slot: CallConcurrencySlot, workflow_run_id: int + ) -> None: + stored = await rate_limiter.store_workflow_slot_mapping_if_absent( + workflow_run_id, + slot.organization_id, + slot.slot_id, + ) + if stored: + return + + await self.release_slot(slot) + raise WorkflowRunSlotAlreadyBoundError(workflow_run_id) + + async def register_active_call( + self, + organization_id: int, + workflow_run_id: int, + *, + source: str, + timeout: float = 0, + max_concurrent_override: int | None = None, + retry_interval: float = 1, + ) -> CallConcurrencySlot: + slot = await self.acquire_org_slot( + organization_id, + source=source, + timeout=timeout, + max_concurrent_override=max_concurrent_override, + retry_interval=retry_interval, + ) + await self.bind_workflow_run(slot, workflow_run_id) + return slot + + async def unregister_active_call(self, workflow_run_id: int) -> bool: + return await self.release_workflow_run_slot(workflow_run_id) + + async def release_slot(self, slot: CallConcurrencySlot | None) -> bool: + if slot is None: + return False + return await rate_limiter.release_concurrent_slot( + slot.organization_id, slot.slot_id + ) + + async def release_workflow_run_slot(self, workflow_run_id: int) -> bool: + mapping = await rate_limiter.get_workflow_slot_mapping(workflow_run_id) + if not mapping: + return False + + org_id, slot_id = mapping + released = await rate_limiter.release_concurrent_slot(org_id, slot_id) + await rate_limiter.delete_workflow_slot_mapping(workflow_run_id) + if released: + logger.info(f"Released concurrent slot for workflow run {workflow_run_id}") + else: + logger.debug( + f"Concurrent slot mapping for workflow run {workflow_run_id} " + "had no live slot; deleted stale mapping" + ) + return released + + +call_concurrency = CallConcurrencyService() diff --git a/api/services/campaign/campaign_call_dispatcher.py b/api/services/campaign/campaign_call_dispatcher.py index 84a419be..f06270d1 100644 --- a/api/services/campaign/campaign_call_dispatcher.py +++ b/api/services/campaign/campaign_call_dispatcher.py @@ -5,10 +5,14 @@ from typing import TYPE_CHECKING, Optional from loguru import logger -from api.constants import DEFAULT_ORG_CONCURRENCY_LIMIT from api.db import db_client from api.db.models import QueuedRunModel, WorkflowRunModel -from api.enums import OrganizationConfigurationKey, WorkflowRunState +from api.enums import WorkflowRunState +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + CallConcurrencySlot, + call_concurrency, +) from api.services.campaign.circuit_breaker import circuit_breaker from api.services.campaign.errors import ( ConcurrentSlotAcquisitionError, @@ -29,9 +33,6 @@ if TYPE_CHECKING: class CampaignCallDispatcher: """Manages rate-limited and concurrent-limited call dispatching""" - def __init__(self): - self.default_concurrent_limit = int(DEFAULT_ORG_CONCURRENCY_LIMIT) - async def get_provider_for_campaign(self, campaign) -> "TelephonyProvider": """Get the telephony provider pinned to this campaign's config. Falls back to the org's default config for legacy campaigns whose @@ -53,18 +54,7 @@ class CampaignCallDispatcher: async def get_org_concurrent_limit(self, organization_id: int) -> int: """Get the concurrent call limit for an organization.""" - try: - config = await db_client.get_configuration( - organization_id, - OrganizationConfigurationKey.CONCURRENT_CALL_LIMIT.value, - ) - if config and config.value: - return int(config.value["value"]) - except Exception as e: - logger.warning( - f"Error getting concurrent limit for org {organization_id}: {e}" - ) - return self.default_concurrent_limit + return await call_concurrency.get_org_concurrent_limit(organization_id) async def process_batch(self, campaign_id: int, batch_size: int = 10) -> int: """ @@ -119,12 +109,14 @@ class CampaignCallDispatcher: ) # Acquire concurrent slot - waits until a slot is available - slot_id = await self.acquire_concurrent_slot( + concurrency_slot = await self.acquire_concurrent_slot( campaign.organization_id, campaign ) # Dispatch the call - workflow_run = await self.dispatch_call(queued_run, campaign, slot_id) + workflow_run = await self.dispatch_call( + queued_run, campaign, concurrency_slot + ) # Update queued run as processed await db_client.update_queued_run( @@ -233,68 +225,61 @@ class CampaignCallDispatcher: ) async def dispatch_call( - self, queued_run: QueuedRunModel, campaign: any, slot_id: str + self, + queued_run: QueuedRunModel, + campaign: any, + concurrency_slot: CallConcurrencySlot, ) -> Optional[WorkflowRunModel]: - """Creates workflow run and initiates call. Requires a pre-acquired slot_id.""" + """Creates workflow run and initiates call. Requires a pre-acquired slot.""" from_number = None + workflow_run = None + slot_bound = False - # Get workflow details - workflow = await db_client.get_workflow_by_id(campaign.workflow_id) - if not workflow: - # Release slot before raising - await rate_limiter.release_concurrent_slot( - campaign.organization_id, slot_id - ) - raise ValueError(f"Workflow {campaign.workflow_id} not found") - - # Extract phone number - phone_number = queued_run.context_variables.get("phone_number") - if not phone_number: - # Release slot before raising - await rate_limiter.release_concurrent_slot( - campaign.organization_id, slot_id - ) - raise ValueError(f"No phone number in queued run {queued_run.id}") - - # Get provider for this campaign's pinned telephony config. - provider = await self.get_provider_for_campaign(campaign) - workflow_run_mode = provider.PROVIDER_NAME - - # Acquire a unique from_number from the pool scoped to this campaign's - # telephony configuration so orgs with multiple configs don't leak - # caller IDs across configs. - from_number = await self.acquire_from_number( - campaign.organization_id, - telephony_configuration_id=campaign.telephony_configuration_id, - ) - if from_number is None: - # Release concurrent slot before raising - await rate_limiter.release_concurrent_slot( - campaign.organization_id, slot_id - ) - raise PhoneNumberPoolExhaustedError( - organization_id=campaign.organization_id - ) - - logger.info(f"Provider name: {provider.PROVIDER_NAME}") - logger.info(f"Queued run context: {queued_run.context_variables}") - - # Merge context variables (queued_run context already includes retry info if applicable) - initial_context = { - **queued_run.context_variables, - "campaign_id": campaign.id, - "provider": provider.PROVIDER_NAME, - "source_uuid": queued_run.source_uuid, - "caller_number": from_number, - "called_number": phone_number, - "telephony_configuration_id": campaign.telephony_configuration_id, - } - - logger.info(f"Final initial_context: {initial_context}") - - # Create workflow run with queued_run_id tracking - workflow_run_name = f"WR-CAMPAIGN-{campaign.id}-{queued_run.id}" try: + # Get workflow details + workflow = await db_client.get_workflow_by_id(campaign.workflow_id) + if not workflow: + raise ValueError(f"Workflow {campaign.workflow_id} not found") + + # Extract phone number + phone_number = queued_run.context_variables.get("phone_number") + if not phone_number: + raise ValueError(f"No phone number in queued run {queued_run.id}") + + # Get provider for this campaign's pinned telephony config. + provider = await self.get_provider_for_campaign(campaign) + workflow_run_mode = provider.PROVIDER_NAME + + # Acquire a unique from_number from the pool scoped to this campaign's + # telephony configuration so orgs with multiple configs don't leak + # caller IDs across configs. + from_number = await self.acquire_from_number( + campaign.organization_id, + telephony_configuration_id=campaign.telephony_configuration_id, + ) + if from_number is None: + raise PhoneNumberPoolExhaustedError( + organization_id=campaign.organization_id + ) + + logger.info(f"Provider name: {provider.PROVIDER_NAME}") + logger.info(f"Queued run context: {queued_run.context_variables}") + + # Merge context variables (queued_run context already includes retry info if applicable) + initial_context = { + **queued_run.context_variables, + "campaign_id": campaign.id, + "provider": provider.PROVIDER_NAME, + "source_uuid": queued_run.source_uuid, + "caller_number": from_number, + "called_number": phone_number, + "telephony_configuration_id": campaign.telephony_configuration_id, + } + + logger.info(f"Final initial_context: {initial_context}") + + # Create workflow run with queued_run_id tracking + workflow_run_name = f"WR-CAMPAIGN-{campaign.id}-{queued_run.id}" workflow_run = await db_client.create_workflow_run( name=workflow_run_name, workflow_id=campaign.workflow_id, @@ -304,11 +289,8 @@ class CampaignCallDispatcher: campaign_id=campaign.id, queued_run_id=queued_run.id, # Link to queued run for retry tracking ) - - # Store slot_id mapping in Redis for cleanup later - await rate_limiter.store_workflow_slot_mapping( - workflow_run.id, campaign.organization_id, slot_id - ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) + slot_bound = True # Store from_number mapping for cleanup on call completion await rate_limiter.store_workflow_from_number_mapping( @@ -319,9 +301,10 @@ class CampaignCallDispatcher: ) except Exception as e: # Release slot and from_number on error - await rate_limiter.release_concurrent_slot( - campaign.organization_id, slot_id - ) + if slot_bound and workflow_run: + await call_concurrency.release_workflow_run_slot(workflow_run.id) + else: + await call_concurrency.release_slot(concurrency_slot) if from_number: await rate_limiter.release_from_number( campaign.organization_id, @@ -357,21 +340,7 @@ class CampaignCallDispatcher: gathered_context={"error": error_message}, ) - mapping = await rate_limiter.get_workflow_slot_mapping(workflow_run.id) - if mapping: - org_id, mapped_slot_id = mapping - await rate_limiter.release_concurrent_slot(org_id, mapped_slot_id) - await rate_limiter.delete_workflow_slot_mapping(workflow_run.id) - - from_number_mapping = await rate_limiter.get_workflow_from_number_mapping( - workflow_run.id - ) - if from_number_mapping: - fn_org_id, fn_number, fn_tcid = from_number_mapping - await rate_limiter.release_from_number( - fn_org_id, fn_number, telephony_configuration_id=fn_tcid - ) - await rate_limiter.delete_workflow_from_number_mapping(workflow_run.id) + await self.release_call_slot(workflow_run.id) raise ValueError(error_message) @@ -446,23 +415,7 @@ class CampaignCallDispatcher: reason="call_initiation_failed", ) - # Release concurrent slot on failure - mapping = await rate_limiter.get_workflow_slot_mapping(workflow_run.id) - if mapping: - org_id, slot_id = mapping - await rate_limiter.release_concurrent_slot(org_id, slot_id) - await rate_limiter.delete_workflow_slot_mapping(workflow_run.id) - - # Release from_number on failure - from_number_mapping = await rate_limiter.get_workflow_from_number_mapping( - workflow_run.id - ) - if from_number_mapping: - fn_org_id, fn_number, fn_tcid = from_number_mapping - await rate_limiter.release_from_number( - fn_org_id, fn_number, telephony_configuration_id=fn_tcid - ) - await rate_limiter.delete_workflow_from_number_mapping(workflow_run.id) + await self.release_call_slot(workflow_run.id) raise @@ -501,7 +454,7 @@ class CampaignCallDispatcher: async def acquire_concurrent_slot( self, organization_id: int, campaign: any, timeout: float = 600 - ) -> str: + ) -> CallConcurrencySlot: """ Acquires a concurrent call slot - waits if necessary until a slot is available. @@ -510,14 +463,11 @@ class CampaignCallDispatcher: campaign: The campaign object timeout: Maximum time to wait for a slot (default 10 minutes) - Returns the slot_id which must be released when the call completes. + Returns the slot which must be released when the call completes. Raises: ConcurrentSlotAcquisitionError: If slot cannot be acquired within timeout """ - # Get concurrent limit for organization - org_concurrent_limit = await self.get_org_concurrent_limit(organization_id) - # Check for campaign-level max_concurrency in orchestrator_metadata campaign_max_concurrency = None if campaign.orchestrator_metadata: @@ -525,39 +475,20 @@ class CampaignCallDispatcher: "max_concurrency" ) - # Use the lower of campaign limit and org limit - if campaign_max_concurrency is not None: - max_concurrent = min(campaign_max_concurrency, org_concurrent_limit) - else: - max_concurrent = org_concurrent_limit - - # Track wait time for alerting - wait_start = time.time() - - # Wait until we can acquire a concurrent slot - while True: - slot_id = await rate_limiter.try_acquire_concurrent_slot( - organization_id, max_concurrent + try: + return await call_concurrency.acquire_org_slot( + organization_id, + source=f"campaign:{campaign.id}", + timeout=timeout, + max_concurrent_override=campaign_max_concurrency, + retry_interval=1, ) - if slot_id: - return slot_id - - # Check if we've been waiting too long - wait_time = time.time() - wait_start - if wait_time > timeout: - raise ConcurrentSlotAcquisitionError( - organization_id=organization_id, - campaign_id=campaign.id, - wait_time=wait_time, - ) - - logger.debug( - f"Attempting to get a slot for {organization_id} {campaign.id}, " - f"waited {wait_time:.1f}s" - ) - - # Wait before retrying - await asyncio.sleep(1) + except CallConcurrencyLimitError as e: + raise ConcurrentSlotAcquisitionError( + organization_id=organization_id, + campaign_id=campaign.id, + wait_time=e.wait_time, + ) from e async def acquire_from_number( self, @@ -602,17 +533,9 @@ class CampaignCallDispatcher: Release concurrent slot and from_number when a call completes. Called by Twilio webhooks or workflow completion handlers. """ - slot_released = False - mapping = await rate_limiter.get_workflow_slot_mapping(workflow_run_id) - if mapping: - org_id, slot_id = mapping - success = await rate_limiter.release_concurrent_slot(org_id, slot_id) - if success: - await rate_limiter.delete_workflow_slot_mapping(workflow_run_id) - logger.info( - f"Released concurrent slot for workflow run {workflow_run_id}" - ) - slot_released = True + slot_released = await call_concurrency.release_workflow_run_slot( + workflow_run_id + ) # Release from_number back to its (org, telephony config) pool from_number_mapping = await rate_limiter.get_workflow_from_number_mapping( diff --git a/api/services/campaign/rate_limiter.py b/api/services/campaign/rate_limiter.py index e9293307..834c09a1 100644 --- a/api/services/campaign/rate_limiter.py +++ b/api/services/campaign/rate_limiter.py @@ -1,5 +1,6 @@ import time import uuid +from dataclasses import dataclass from typing import Optional import redis.asyncio as aioredis @@ -8,6 +9,12 @@ from loguru import logger from api.constants import REDIS_URL +@dataclass(frozen=True) +class ConcurrentSlotAcquisition: + slot_id: str + active_count: int + + class RateLimiter: """Sliding window rate limiter to enforce strict per-second limits and concurrent call limits""" @@ -100,6 +107,19 @@ class RateLimiter: Try to acquire a concurrent call slot. Returns a unique slot_id if successful, None if limit reached. """ + acquisition = await self.try_acquire_concurrent_slot_details( + organization_id, max_concurrent + ) + return acquisition.slot_id if acquisition else None + + async def try_acquire_concurrent_slot_details( + self, organization_id: int, max_concurrent: int = 20 + ) -> Optional[ConcurrentSlotAcquisition]: + """ + Try to acquire a concurrent call slot. + Returns the slot_id and post-acquire active count if successful, + or None if the limit is reached. + """ redis_client = await self._get_redis() concurrent_key = f"concurrent_calls:{organization_id}" @@ -124,7 +144,7 @@ class RateLimiter: -- Add new slot redis.call('ZADD', key, now, slot_id) redis.call('EXPIRE', key, 3600) -- Expire after 1 hour - return slot_id + return {slot_id, current_count + 1} else return nil end @@ -143,7 +163,14 @@ class RateLimiter: stale_cutoff, slot_id, ) - return result + if not result: + return None + + acquired_slot_id, active_count = result + return ConcurrentSlotAcquisition( + slot_id=str(acquired_slot_id), + active_count=int(active_count), + ) except Exception as e: logger.error(f"Concurrent limiter error: {e}") return None @@ -212,6 +239,46 @@ class RateLimiter: logger.error(f"Error storing workflow slot mapping: {e}") return False + async def store_workflow_slot_mapping_if_absent( + self, workflow_run_id: int, organization_id: int, slot_id: str + ) -> bool: + """ + Store the workflow_run_id -> concurrent slot mapping only if no mapping + already exists. This prevents duplicate public/WebRTC starts for the + same workflow run from overwriting the cleanup pointer. + """ + redis_client = await self._get_redis() + mapping_key = f"workflow_slot_mapping:{workflow_run_id}" + + lua_script = """ + local key = KEYS[1] + local org_id = ARGV[1] + local slot_id = ARGV[2] + local ttl = tonumber(ARGV[3]) + + if redis.call('EXISTS', key) == 1 then + return 0 + end + + redis.call('HSET', key, 'org_id', org_id, 'slot_id', slot_id) + redis.call('EXPIRE', key, ttl) + return 1 + """ + + try: + stored = await redis_client.eval( + lua_script, + 1, + mapping_key, + organization_id, + slot_id, + self.stale_call_timeout, + ) + return bool(stored) + except Exception as e: + logger.error(f"Error storing workflow slot mapping if absent: {e}") + return False + async def get_workflow_slot_mapping( self, workflow_run_id: int ) -> Optional[tuple[int, str]]: diff --git a/api/services/pipecat/run_pipeline.py b/api/services/pipecat/run_pipeline.py index 4bfedc7c..44f702d1 100644 --- a/api/services/pipecat/run_pipeline.py +++ b/api/services/pipecat/run_pipeline.py @@ -14,14 +14,17 @@ from api.schemas.workflow_configurations import ( DEFAULT_TURN_START_MIN_WORDS, DEFAULT_TURN_START_STRATEGY, ) +from api.services.call_concurrency import call_concurrency from api.services.configuration.registry import ServiceProviders from api.services.integrations import ( IntegrationRuntimeContext, create_runtime_sessions, ) from api.services.pipecat.active_calls import ( - register_active_call, - unregister_active_call, + register_active_call as register_worker_active_call, +) +from api.services.pipecat.active_calls import ( + unregister_active_call as unregister_worker_active_call, ) from api.services.pipecat.audio_config import AudioConfig, create_audio_config from api.services.pipecat.event_handlers import ( @@ -258,7 +261,7 @@ async def run_pipeline_telephony( """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) + register_worker_active_call(workflow_run_id) try: await _run_pipeline_telephony_impl( websocket, @@ -270,7 +273,10 @@ async def run_pipeline_telephony( transport_kwargs=transport_kwargs, ) finally: - unregister_active_call(workflow_run_id) + try: + await call_concurrency.unregister_active_call(workflow_run_id) + finally: + unregister_worker_active_call(workflow_run_id) async def _run_pipeline_telephony_impl( @@ -383,7 +389,7 @@ async def run_pipeline_smallwebrtc( """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) + register_worker_active_call(workflow_run_id) try: await _run_pipeline_smallwebrtc_impl( webrtc_connection, @@ -394,7 +400,10 @@ async def run_pipeline_smallwebrtc( user_provider_id=user_provider_id, ) finally: - unregister_active_call(workflow_run_id) + try: + await call_concurrency.unregister_active_call(workflow_run_id) + finally: + unregister_worker_active_call(workflow_run_id) async def _run_pipeline_smallwebrtc_impl( @@ -478,7 +487,7 @@ async def _run_pipeline( resolved_user_config=None, ) -> None: """Run the pipeline with active-call drain accounting.""" - register_active_call(workflow_run_id) + register_worker_active_call(workflow_run_id) try: await _run_pipeline_impl( transport, @@ -492,7 +501,10 @@ async def _run_pipeline( resolved_user_config=resolved_user_config, ) finally: - unregister_active_call(workflow_run_id) + try: + await call_concurrency.unregister_active_call(workflow_run_id) + finally: + unregister_worker_active_call(workflow_run_id) async def _run_pipeline_impl( diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index 3663df88..cf8ffffd 100644 --- a/api/services/telephony/ari_manager.py +++ b/api/services/telephony/ari_manager.py @@ -26,6 +26,10 @@ from loguru import logger from api.constants import REDIS_URL from api.db import db_client from api.enums import CallType, WorkflowRunMode +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + call_concurrency, +) from api.services.quota_service import authorize_workflow_run_start from api.services.telephony.call_transfer_manager import get_call_transfer_manager from api.services.telephony.transfer_event_protocol import ( @@ -527,6 +531,8 @@ class ARIConnection: channel = event.get("channel", {}) caller_number = channel.get("caller", {}).get("number", "unknown") called_number = channel.get("dialplan", {}).get("exten", "unknown") + concurrency_slot = None + workflow_run = None try: # 1. Resolve the workflow from the called extension via the @@ -573,6 +579,20 @@ class ARIConnection: user_id = workflow.user_id + try: + concurrency_slot = await call_concurrency.acquire_org_slot( + self.organization_id, + source="ari_inbound", + timeout=0, + ) + except CallConcurrencyLimitError: + logger.warning( + f"[ARI org={self.organization_id}] Concurrent call limit " + f"reached; hanging up inbound channel {channel_id}" + ) + await self._delete_channel(channel_id) + return + # 3. Create workflow run call_id = channel_id workflow_run = await db_client.create_workflow_run( @@ -591,7 +611,9 @@ class ARIConnection: gathered_context={ "call_id": call_id, }, + organization_id=self.organization_id, ) + await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id) logger.info( f"[ARI org={self.organization_id}] Created inbound workflow run " @@ -610,6 +632,7 @@ class ARIConnection: f"[ARI org={self.organization_id}] Quota exceeded for user {user_id} " f"— hanging up inbound call {channel_id}" ) + await call_concurrency.release_workflow_run_slot(workflow_run.id) await self._delete_channel(channel_id) return @@ -625,6 +648,10 @@ class ARIConnection: str(user_id), ) except Exception as e: + if workflow_run: + await call_concurrency.release_workflow_run_slot(workflow_run.id) + elif concurrency_slot: + await call_concurrency.release_slot(concurrency_slot) logger.error( f"[ARI org={self.organization_id}] Error handling inbound StasisStart " f"for channel {channel_id}: {e}" diff --git a/api/services/telephony/status_processor.py b/api/services/telephony/status_processor.py index c5d39b98..f125931c 100644 --- a/api/services/telephony/status_processor.py +++ b/api/services/telephony/status_processor.py @@ -144,8 +144,9 @@ async def _process_status_update(workflow_run_id: int, status: StatusCallbackReq f"[run {workflow_run_id}] Call completed with duration: {status.duration}s" ) + await campaign_call_dispatcher.release_call_slot(workflow_run_id) + if workflow_run.campaign_id: - await campaign_call_dispatcher.release_call_slot(workflow_run_id) await circuit_breaker.record_and_evaluate( workflow_run.campaign_id, is_failure=False ) @@ -162,8 +163,9 @@ async def _process_status_update(workflow_run_id: int, status: StatusCallbackReq f"[run {workflow_run_id}] Call failed with status: {normalized_status.value}" ) + await campaign_call_dispatcher.release_call_slot(workflow_run_id) + if workflow_run.campaign_id: - await campaign_call_dispatcher.release_call_slot(workflow_run_id) is_failure = normalized_status in FAILURE_NOT_CONNECTED_STATUSES await circuit_breaker.record_and_evaluate( workflow_run.campaign_id, diff --git a/api/tests/telephony/test_status_processor.py b/api/tests/telephony/test_status_processor.py index e0bbfd4a..35613723 100644 --- a/api/tests/telephony/test_status_processor.py +++ b/api/tests/telephony/test_status_processor.py @@ -29,6 +29,9 @@ async def test_initialized_no_answer_enqueues_workflow_completion(): with ( patch("api.services.telephony.status_processor.db_client") as mock_db, + patch( + "api.services.telephony.status_processor.campaign_call_dispatcher" + ) as mock_dispatcher, patch( "api.services.telephony.status_processor.enqueue_job", new_callable=AsyncMock, @@ -36,6 +39,7 @@ async def test_initialized_no_answer_enqueues_workflow_completion(): ): mock_db.get_workflow_run_by_id = AsyncMock(return_value=workflow_run) mock_db.update_workflow_run = AsyncMock() + mock_dispatcher.release_call_slot = AsyncMock(return_value=True) await _process_status_update(123, status) @@ -58,6 +62,7 @@ async def test_initialized_no_answer_enqueues_workflow_completion(): mock_enqueue.assert_awaited_once_with( FunctionNames.RUN_INTEGRATIONS_POST_WORKFLOW_RUN, 123 ) + mock_dispatcher.release_call_slot.assert_awaited_once_with(123) @pytest.mark.asyncio @@ -79,6 +84,9 @@ async def test_running_terminal_status_does_not_enqueue_workflow_completion(): with ( patch("api.services.telephony.status_processor.db_client") as mock_db, + patch( + "api.services.telephony.status_processor.campaign_call_dispatcher" + ) as mock_dispatcher, patch( "api.services.telephony.status_processor.enqueue_job", new_callable=AsyncMock, @@ -86,6 +94,7 @@ async def test_running_terminal_status_does_not_enqueue_workflow_completion(): ): mock_db.get_workflow_run_by_id = AsyncMock(return_value=workflow_run) mock_db.update_workflow_run = AsyncMock() + mock_dispatcher.release_call_slot = AsyncMock(return_value=True) await _process_status_update(456, status) @@ -96,3 +105,4 @@ async def test_running_terminal_status_does_not_enqueue_workflow_completion(): "telephony_failed", ] mock_enqueue.assert_not_awaited() + mock_dispatcher.release_call_slot.assert_awaited_once_with(456) diff --git a/api/tests/test_active_calls.py b/api/tests/test_active_calls.py index 40447906..575eb046 100644 --- a/api/tests/test_active_calls.py +++ b/api/tests/test_active_calls.py @@ -7,6 +7,7 @@ idempotent, and the count only reaches zero when every registered run is gone. """ import asyncio +from unittest.mock import AsyncMock import pytest from fastapi import FastAPI @@ -91,6 +92,12 @@ async def test_run_pipeline_counts_call_during_setup(monkeypatch): "get_workflow_run", fake_get_workflow_run, ) + unregister_concurrency = AsyncMock() + monkeypatch.setattr( + run_pipeline_module.call_concurrency, + "unregister_active_call", + unregister_concurrency, + ) task = asyncio.create_task( run_pipeline_module._run_pipeline( @@ -108,6 +115,7 @@ async def test_run_pipeline_counts_call_during_setup(monkeypatch): with pytest.raises(RuntimeError, match="setup failed"): await asyncio.wait_for(task, timeout=1.0) assert active_calls.active_call_count() == 0 + unregister_concurrency.assert_awaited_once_with(42) @pytest.mark.asyncio @@ -123,6 +131,12 @@ async def test_webrtc_entrypoint_counts_call_during_setup(monkeypatch): monkeypatch.setattr( run_pipeline_module.db_client, "get_workflow", fake_get_workflow ) + unregister_concurrency = AsyncMock() + monkeypatch.setattr( + run_pipeline_module.call_concurrency, + "unregister_active_call", + unregister_concurrency, + ) task = asyncio.create_task( run_pipeline_module.run_pipeline_smallwebrtc( @@ -140,6 +154,7 @@ async def test_webrtc_entrypoint_counts_call_during_setup(monkeypatch): with pytest.raises(RuntimeError, match="setup failed"): await asyncio.wait_for(task, timeout=1.0) assert active_calls.active_call_count() == 0 + unregister_concurrency.assert_awaited_once_with(43) @pytest.mark.asyncio @@ -155,6 +170,12 @@ async def test_telephony_entrypoint_counts_call_during_setup(monkeypatch): monkeypatch.setattr( run_pipeline_module.db_client, "get_workflow", fake_get_workflow ) + unregister_concurrency = AsyncMock() + monkeypatch.setattr( + run_pipeline_module.call_concurrency, + "unregister_active_call", + unregister_concurrency, + ) task = asyncio.create_task( run_pipeline_module.run_pipeline_telephony( @@ -175,6 +196,7 @@ async def test_telephony_entrypoint_counts_call_during_setup(monkeypatch): with pytest.raises(RuntimeError, match="setup failed"): await asyncio.wait_for(task, timeout=1.0) assert active_calls.active_call_count() == 0 + unregister_concurrency.assert_awaited_once_with(44) def test_active_calls_route_requires_configured_secret(monkeypatch): diff --git a/api/tests/test_agent_stream_route.py b/api/tests/test_agent_stream_route.py index b789aa5c..5a143c07 100644 --- a/api/tests/test_agent_stream_route.py +++ b/api/tests/test_agent_stream_route.py @@ -3,6 +3,8 @@ from unittest.mock import AsyncMock, patch import pytest +from api.services.call_concurrency import CallConcurrencyLimitError + class _FakeWebSocket: def __init__(self, query_params: dict[str, str] | None = None): @@ -34,6 +36,7 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): with ( patch("api.routes.agent_stream.telephony_registry") as registry, patch("api.routes.agent_stream.db_client") as db_client, + patch("api.routes.agent_stream.call_concurrency") as mock_concurrency, patch( "api.routes.agent_stream.authorize_workflow_run_start", new=AsyncMock( @@ -41,6 +44,13 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): ), ), ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.unregister_active_call = AsyncMock() + registry.get_optional.return_value = spec db_client.get_workflow_by_uuid_unscoped = AsyncMock(return_value=workflow) db_client.create_workflow_run = AsyncMock(return_value=workflow_run) @@ -50,9 +60,17 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): registry.get_optional.assert_called_once_with("cloudonix") db_client.create_workflow_run.assert_awaited_once() + mock_concurrency.acquire_org_slot.assert_awaited_once_with( + workflow.organization_id, + source="agent_stream:cloudonix", + timeout=0, + ) + mock_concurrency.bind_workflow_run.assert_awaited_once_with(slot, workflow_run.id) + mock_concurrency.unregister_active_call.assert_awaited_once_with(workflow_run.id) create_args = db_client.create_workflow_run.await_args.args create_kwargs = db_client.create_workflow_run.await_args.kwargs assert create_args[2] == "cloudonix" + assert create_kwargs["organization_id"] == workflow.organization_id assert create_kwargs["initial_context"] == { "existing": "context", "provider": "cloudonix", @@ -62,3 +80,42 @@ async def test_agent_stream_uses_provider_path_param_not_query_param(): _, provider_kwargs = provider.handle_external_websocket.await_args assert provider_kwargs["params"] == {"custom": "value"} websocket.close.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_agent_stream_rejects_when_concurrency_limit_reached(): + from api.routes.agent_stream import agent_stream_websocket + + websocket = _FakeWebSocket() + workflow = SimpleNamespace( + id=11, + user_id=22, + organization_id=33, + template_context_variables={}, + ) + spec = SimpleNamespace(provider_cls=lambda _config: object()) + + with ( + patch("api.routes.agent_stream.telephony_registry") as registry, + patch("api.routes.agent_stream.db_client") as db_client, + patch("api.routes.agent_stream.call_concurrency") as mock_concurrency, + ): + registry.get_optional.return_value = spec + db_client.get_workflow_by_uuid_unscoped = AsyncMock(return_value=workflow) + db_client.create_workflow_run = AsyncMock() + mock_concurrency.acquire_org_slot = AsyncMock( + side_effect=CallConcurrencyLimitError( + organization_id=workflow.organization_id, + source="agent_stream:cloudonix", + wait_time=0, + max_concurrent=1, + ) + ) + + await agent_stream_websocket(websocket, "cloudonix", "agent-uuid") + + websocket.close.assert_awaited_once_with( + code=1008, + reason="Concurrent call limit reached", + ) + db_client.create_workflow_run.assert_not_awaited() diff --git a/api/tests/test_call_concurrency.py b/api/tests/test_call_concurrency.py new file mode 100644 index 00000000..cb65ee31 --- /dev/null +++ b/api/tests/test_call_concurrency.py @@ -0,0 +1,69 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from api.services.call_concurrency import ( + CallConcurrencyLimitError, + CallConcurrencyService, +) +from api.services.campaign.rate_limiter import ConcurrentSlotAcquisition + + +@pytest.mark.asyncio +async def test_acquire_org_slot_logs_post_acquire_count_and_limit(): + service = CallConcurrencyService() + + with ( + patch("api.services.call_concurrency.db_client") as mock_db, + patch("api.services.call_concurrency.rate_limiter") as mock_rate_limiter, + patch("api.services.call_concurrency.logger") as mock_logger, + ): + mock_db.get_configuration = AsyncMock(return_value=None) + mock_rate_limiter.try_acquire_concurrent_slot_details = AsyncMock( + return_value=ConcurrentSlotAcquisition( + slot_id="slot-123", + active_count=7, + ) + ) + + slot = await service.acquire_org_slot(199, source="test_source") + + assert slot.organization_id == 199 + assert slot.slot_id == "slot-123" + assert slot.max_concurrent == 10 + assert slot.source == "test_source" + mock_rate_limiter.try_acquire_concurrent_slot_details.assert_awaited_once_with( + 199, 10 + ) + mock_logger.info.assert_called_once() + log_message = mock_logger.info.call_args.args[0] + assert "org 199" in log_message + assert "source=test_source" in log_message + assert "active_calls=7/10" in log_message + assert "slot_id=slot-123" in log_message + + +@pytest.mark.asyncio +async def test_acquire_org_slot_logs_warning_when_limit_reached(): + service = CallConcurrencyService() + + with ( + patch("api.services.call_concurrency.db_client") as mock_db, + patch("api.services.call_concurrency.rate_limiter") as mock_rate_limiter, + patch("api.services.call_concurrency.logger") as mock_logger, + ): + mock_db.get_configuration = AsyncMock(return_value=None) + mock_rate_limiter.try_acquire_concurrent_slot_details = AsyncMock( + return_value=None + ) + mock_rate_limiter.get_concurrent_count = AsyncMock(return_value=12) + + with pytest.raises(CallConcurrencyLimitError): + await service.acquire_org_slot(199, source="test_source", timeout=0) + + mock_rate_limiter.get_concurrent_count.assert_awaited_once_with(199) + mock_logger.warning.assert_called_once() + log_message = mock_logger.warning.call_args.args[0] + assert "Concurrent call limit reached for org 199" in log_message + assert "source=test_source" in log_message + assert "active_calls=12/10" in log_message diff --git a/api/tests/test_campaign_call_dispatcher.py b/api/tests/test_campaign_call_dispatcher.py index f609cec1..2ec33289 100644 --- a/api/tests/test_campaign_call_dispatcher.py +++ b/api/tests/test_campaign_call_dispatcher.py @@ -25,6 +25,7 @@ from api.db.models import ( WorkflowModel, WorkflowRunModel, ) +from api.services.call_concurrency import CallConcurrencySlot from api.services.campaign.campaign_call_dispatcher import CampaignCallDispatcher # ============================================================================= @@ -259,6 +260,26 @@ def mock_rate_limiter(): } +@pytest.fixture(autouse=True) +def mock_call_concurrency(): + async def acquire_slot(organization_id, *, source, **kwargs): + return CallConcurrencySlot( + organization_id=organization_id, + slot_id=f"slot-{uuid.uuid4().hex[:8]}", + max_concurrent=kwargs.get("max_concurrent_override") or 20, + source=source, + ) + + with patch( + "api.services.campaign.campaign_call_dispatcher.call_concurrency" + ) as mock_concurrency: + mock_concurrency.acquire_org_slot = AsyncMock(side_effect=acquire_slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock(return_value=True) + mock_concurrency.release_workflow_run_slot = AsyncMock(return_value=True) + yield mock_concurrency + + # ============================================================================= # Tests # ============================================================================= diff --git a/api/tests/test_circuit_breaker.py b/api/tests/test_circuit_breaker.py index 8d975841..9b5e61c1 100644 --- a/api/tests/test_circuit_breaker.py +++ b/api/tests/test_circuit_breaker.py @@ -677,15 +677,20 @@ class TestProcessStatusUpdateCircuitBreaker: with ( patch("api.services.telephony.status_processor.db_client") as mock_db, + patch( + "api.services.telephony.status_processor.campaign_call_dispatcher" + ) as mock_dispatcher, patch("api.services.telephony.status_processor.circuit_breaker") as mock_cb, ): mock_db.get_workflow_run_by_id = AsyncMock(return_value=mock_workflow_run) mock_db.update_workflow_run = AsyncMock() + mock_dispatcher.release_call_slot = AsyncMock(return_value=True) await _process_status_update(100, status) # Circuit breaker should NOT be called for non-campaign calls mock_cb.record_and_evaluate.assert_not_called() + mock_dispatcher.release_call_slot.assert_awaited_once_with(100) # ============================================================================= diff --git a/api/tests/test_from_number_pool_isolation.py b/api/tests/test_from_number_pool_isolation.py index ae3dffbc..5e393111 100644 --- a/api/tests/test_from_number_pool_isolation.py +++ b/api/tests/test_from_number_pool_isolation.py @@ -20,6 +20,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest +from api.services.call_concurrency import CallConcurrencySlot from api.services.campaign.campaign_call_dispatcher import CampaignCallDispatcher from api.services.campaign.rate_limiter import RateLimiter @@ -266,6 +267,9 @@ class TestDispatcherThreadsTelephonyConfig: patch( "api.services.campaign.campaign_call_dispatcher.rate_limiter" ) as mock_rl, + patch( + "api.services.campaign.campaign_call_dispatcher.call_concurrency" + ) as mock_concurrency, patch( "api.services.campaign.campaign_call_dispatcher.get_backend_endpoints", AsyncMock(return_value=("https://example.com", None)), @@ -280,14 +284,21 @@ class TestDispatcherThreadsTelephonyConfig: mock_db.get_workflow_by_id = AsyncMock(return_value=SimpleNamespace(id=1)) mock_db.create_workflow_run = AsyncMock(return_value=workflow_run) mock_db.update_workflow_run = AsyncMock() + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() mock_rl.acquire_from_number = AsyncMock(return_value="+15551110001") mock_rl.release_from_number = AsyncMock() - mock_rl.release_concurrent_slot = AsyncMock() - mock_rl.store_workflow_slot_mapping = AsyncMock() mock_rl.store_workflow_from_number_mapping = AsyncMock() - await dispatcher.dispatch_call(queued_run, campaign, slot_id="slot-1") + slot = CallConcurrencySlot( + organization_id=org_id, + slot_id="slot-1", + max_concurrent=1, + source="test", + ) + await dispatcher.dispatch_call(queued_run, campaign, slot) # acquire_from_number on rate_limiter must be called with the # campaign's telephony_configuration_id. @@ -337,10 +348,15 @@ class TestDispatcherThreadsTelephonyConfig: dispatcher = CampaignCallDispatcher() - with patch( - "api.services.campaign.campaign_call_dispatcher.rate_limiter" - ) as mock_rl: - mock_rl.get_workflow_slot_mapping = AsyncMock(return_value=None) + with ( + patch( + "api.services.campaign.campaign_call_dispatcher.rate_limiter" + ) as mock_rl, + patch( + "api.services.campaign.campaign_call_dispatcher.call_concurrency" + ) as mock_concurrency, + ): + mock_concurrency.release_workflow_run_slot = AsyncMock(return_value=False) mock_rl.get_workflow_from_number_mapping = AsyncMock( return_value=(org_id, from_number, config_id) ) diff --git a/api/tests/test_public_agent_routes.py b/api/tests/test_public_agent_routes.py index 3b7ea409..3eb9412f 100644 --- a/api/tests/test_public_agent_routes.py +++ b/api/tests/test_public_agent_routes.py @@ -5,6 +5,7 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from api.routes.public_agent import router +from api.services.call_concurrency import CallConcurrencyLimitError def _make_test_app() -> FastAPI: @@ -56,6 +57,7 @@ def test_trigger_route_executes_as_workflow_owner(): with ( patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, patch( "api.routes.public_agent.authorize_workflow_run_start", new=quota_mock, @@ -69,6 +71,12 @@ def test_trigger_route_executes_as_workflow_owner(): new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), ), ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_db.validate_api_key = AsyncMock( return_value=SimpleNamespace(id=7, organization_id=11, created_by=22) ) @@ -96,6 +104,12 @@ def test_trigger_route_executes_as_workflow_owner(): workflow_id=workflow.id, workflow_run_id=501, ) + mock_concurrency.acquire_org_slot.assert_awaited_once_with( + workflow.organization_id, + source="public_agent", + timeout=0, + ) + mock_concurrency.bind_workflow_run.assert_awaited_once_with(slot, 501) mock_db.get_workflow.assert_awaited_once_with(workflow.id, organization_id=11) create_kwargs = mock_db.create_workflow_run.await_args.kwargs @@ -126,6 +140,7 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution(): with ( patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, patch( "api.routes.public_agent.authorize_workflow_run_start", new=quota_mock, @@ -139,6 +154,11 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution(): new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), ), ): + mock_concurrency.acquire_org_slot = AsyncMock(return_value=object()) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_db.validate_api_key = AsyncMock( return_value=SimpleNamespace(id=8, organization_id=11, created_by=22) ) @@ -170,6 +190,110 @@ def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution(): assert "agent_uuid" not in create_kwargs["initial_context"] +def test_trigger_route_rejects_when_concurrency_limit_reached(): + app = _make_test_app() + client = TestClient(app) + + workflow = _active_workflow(trigger_path="trigger-uuid-123") + provider = _provider() + + with ( + patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, + patch( + "api.routes.public_agent.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + ): + mock_concurrency.acquire_org_slot = AsyncMock( + side_effect=CallConcurrencyLimitError( + organization_id=11, + source="public_agent", + wait_time=0, + max_concurrent=2, + ) + ) + mock_db.validate_api_key = AsyncMock( + return_value=SimpleNamespace(id=7, organization_id=11, created_by=22) + ) + mock_db.get_agent_trigger_by_path = AsyncMock( + return_value=SimpleNamespace( + workflow_id=workflow.id, + organization_id=11, + state="active", + ) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.create_workflow_run = AsyncMock() + + response = client.post( + "/public/agent/trigger-uuid-123", + headers={"X-API-Key": "test-api-key"}, + json={"phone_number": "+15551234567"}, + ) + + assert response.status_code == 429 + assert response.json()["detail"] == "Concurrent call limit reached" + mock_db.create_workflow_run.assert_not_called() + + +def test_trigger_route_releases_concurrency_slot_when_quota_fails(): + app = _make_test_app() + client = TestClient(app) + + workflow = _active_workflow(trigger_path="trigger-uuid-123") + provider = _provider() + quota_mock = AsyncMock( + return_value=SimpleNamespace(has_quota=False, error_message="Quota exceeded") + ) + + with ( + patch("api.routes.public_agent.db_client") as mock_db, + patch("api.routes.public_agent.call_concurrency") as mock_concurrency, + patch( + "api.routes.public_agent.authorize_workflow_run_start", + new=quota_mock, + ), + patch( + "api.routes.public_agent.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + ): + mock_concurrency.acquire_org_slot = AsyncMock(return_value=object()) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + + mock_db.validate_api_key = AsyncMock( + return_value=SimpleNamespace(id=7, organization_id=11, created_by=22) + ) + mock_db.get_agent_trigger_by_path = AsyncMock( + return_value=SimpleNamespace( + workflow_id=workflow.id, + organization_id=11, + state="active", + ) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.create_workflow_run = AsyncMock(return_value=SimpleNamespace(id=501)) + + response = client.post( + "/public/agent/trigger-uuid-123", + headers={"X-API-Key": "test-api-key"}, + json={"phone_number": "+15551234567"}, + ) + + assert response.status_code == 402 + mock_concurrency.release_workflow_run_slot.assert_awaited_once_with(501) + provider.initiate_call.assert_not_awaited() + + def test_workflow_uuid_route_rejects_archived_workflows(): app = _make_test_app() client = TestClient(app) diff --git a/api/tests/test_public_embed_cors.py b/api/tests/test_public_embed_cors.py index 5683f38c..1608fc72 100644 --- a/api/tests/test_public_embed_cors.py +++ b/api/tests/test_public_embed_cors.py @@ -25,6 +25,7 @@ _ACTIVE_TOKEN = SimpleNamespace( expires_at=None, allowed_domains=[], workflow_id=1, + organization_id=11, created_by=7, usage_limit=None, usage_count=0, @@ -37,6 +38,7 @@ _RESTRICTED_TOKEN = SimpleNamespace( expires_at=None, allowed_domains=["allowed.example.com"], workflow_id=2, + organization_id=11, created_by=7, usage_limit=None, usage_count=0, @@ -49,6 +51,7 @@ _LOCALHOST_TOKEN = SimpleNamespace( expires_at=None, allowed_domains=["localhost:3000", "localhost:3020"], workflow_id=3, + organization_id=11, created_by=7, usage_limit=None, usage_count=0, diff --git a/api/tests/test_telephony_routes.py b/api/tests/test_telephony_routes.py index 1a8157d6..24d8a7d8 100644 --- a/api/tests/test_telephony_routes.py +++ b/api/tests/test_telephony_routes.py @@ -6,8 +6,10 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from api.enums import WorkflowRunMode, WorkflowRunState -from api.routes.telephony import _handle_telephony_websocket, router +from api.errors.telephony_errors import TelephonyError +from api.routes.telephony import _handle_telephony_websocket, handle_inbound_run, router from api.services.auth.depends import get_user +from api.services.call_concurrency import CallConcurrencyLimitError def _make_test_app() -> FastAPI: @@ -55,6 +57,7 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): with ( patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, patch( "api.routes.telephony.authorize_workflow_run_start", new=quota_mock, @@ -68,6 +71,12 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), ), ): + slot = object() + mock_concurrency.acquire_org_slot = AsyncMock(return_value=slot) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_db.get_user_configurations = AsyncMock( return_value=SimpleNamespace(test_phone_number=None) ) @@ -104,6 +113,12 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): assert create_kwargs["user_id"] == workflow.user_id assert create_kwargs["organization_id"] == workflow.organization_id assert create_kwargs["initial_context"]["template_key"] == "template-value" + mock_concurrency.acquire_org_slot.assert_awaited_once_with( + workflow.organization_id, + source="telephony_outbound", + timeout=0, + ) + mock_concurrency.bind_workflow_run.assert_awaited_once_with(slot, 501) initiate_kwargs = provider.initiate_call.await_args.kwargs assert initiate_kwargs["workflow_id"] == workflow.id @@ -124,6 +139,7 @@ def test_initiate_call_uses_organization_preference_phone_number(): with ( patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, patch( "api.routes.telephony.authorize_workflow_run_start", new=quota_mock, @@ -137,6 +153,11 @@ def test_initiate_call_uses_organization_preference_phone_number(): new=AsyncMock(return_value=("https://api.example.com", "wss://ignored")), ), ): + mock_concurrency.acquire_org_slot = AsyncMock(return_value=object()) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_db.get_user_configurations = AsyncMock( return_value=SimpleNamespace(test_phone_number="+15550000000") ) @@ -178,6 +199,7 @@ def test_initiate_call_rejects_existing_run_for_different_workflow(): with ( patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, patch( "api.routes.telephony.authorize_workflow_run_start", new=quota_mock, @@ -187,6 +209,11 @@ def test_initiate_call_rejects_existing_run_for_different_workflow(): new=AsyncMock(return_value=provider), ), ): + mock_concurrency.acquire_org_slot = AsyncMock(return_value=object()) + mock_concurrency.bind_workflow_run = AsyncMock() + mock_concurrency.release_slot = AsyncMock() + mock_concurrency.release_workflow_run_slot = AsyncMock() + mock_db.get_user_configurations = AsyncMock( return_value=SimpleNamespace(test_phone_number=None) ) @@ -215,10 +242,119 @@ def test_initiate_call_rejects_existing_run_for_different_workflow(): assert response.status_code == 400 assert response.json()["detail"] == "workflow_run_workflow_mismatch" mock_db.get_workflow_run.assert_awaited_once_with(501, organization_id=11) + mock_concurrency.release_slot.assert_awaited_once() assert not mock_db.create_workflow_run.called assert provider.initiate_call.await_count == 0 +def test_initiate_call_rejects_when_concurrency_limit_reached(): + app = _make_test_app() + client = TestClient(app) + + workflow = _workflow() + provider = _provider() + + with ( + patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, + patch( + "api.routes.telephony.get_default_telephony_provider", + new=AsyncMock(return_value=provider), + ), + ): + mock_concurrency.acquire_org_slot = AsyncMock( + side_effect=CallConcurrencyLimitError( + organization_id=workflow.organization_id, + source="telephony_outbound", + wait_time=0, + max_concurrent=1, + ) + ) + mock_db.get_default_telephony_configuration = AsyncMock( + return_value=SimpleNamespace(id=55) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.create_workflow_run = AsyncMock() + + response = client.post( + "/telephony/initiate-call", + json={"workflow_id": workflow.id, "phone_number": "+15551234567"}, + ) + + assert response.status_code == 429 + assert response.json()["detail"] == "Concurrent call limit reached" + mock_db.create_workflow_run.assert_not_called() + provider.initiate_call.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_inbound_run_rejects_when_concurrency_limit_reached(): + request = SimpleNamespace(headers={}, url="https://api.example.com/inbound/run") + provider_class = SimpleNamespace( + PROVIDER_NAME="twilio", + generate_validation_error_response=Mock(return_value="limit-response"), + ) + normalized_data = SimpleNamespace( + provider="twilio", + direction="inbound", + to_number="+15551230000", + from_number="+15557650000", + to_country="US", + from_country="US", + account_id="acct-1", + call_id="call-1", + raw_data={}, + ) + config = SimpleNamespace(id=55, organization_id=11) + phone_row = SimpleNamespace(id=77, inbound_workflow_id=33) + workflow = SimpleNamespace(id=33, user_id=99) + provider_instance = SimpleNamespace( + verify_inbound_signature=AsyncMock(return_value=True) + ) + + with ( + patch( + "api.routes.telephony.parse_webhook_request", + new=AsyncMock(return_value=({}, "raw-body")), + ), + patch( + "api.routes.telephony._detect_provider", + new=AsyncMock(return_value=provider_class), + ), + patch( + "api.routes.telephony.normalize_webhook_data", + return_value=normalized_data, + ), + patch("api.routes.telephony.db_client") as mock_db, + patch( + "api.routes.telephony.get_telephony_provider_by_id", + new=AsyncMock(return_value=provider_instance), + ), + patch("api.routes.telephony.call_concurrency") as mock_concurrency, + ): + mock_db.find_inbound_route_by_account = AsyncMock( + return_value=(config, phone_row) + ) + mock_db.get_workflow = AsyncMock(return_value=workflow) + mock_db.create_workflow_run = AsyncMock() + mock_concurrency.acquire_org_slot = AsyncMock( + side_effect=CallConcurrencyLimitError( + organization_id=config.organization_id, + source="inbound:twilio", + wait_time=0, + max_concurrent=1, + ) + ) + + response = await handle_inbound_run(request) + + assert response == "limit-response" + provider_class.generate_validation_error_response.assert_called_once_with( + TelephonyError.CONCURRENT_CALL_LIMIT + ) + mock_db.create_workflow_run.assert_not_awaited() + + @pytest.mark.asyncio async def test_smallwebrtc_run_reaching_telephony_websocket_closes_without_running(): websocket = AsyncMock() @@ -235,11 +371,13 @@ async def test_smallwebrtc_run_reaching_telephony_websocket_closes_without_runni with ( patch("api.routes.telephony.db_client") as mock_db, + patch("api.routes.telephony.call_concurrency") as mock_concurrency, patch( "api.routes.telephony.get_telephony_provider_for_run", new=provider_lookup, ), ): + mock_concurrency.unregister_active_call = AsyncMock() mock_db.get_workflow_run = AsyncMock(return_value=workflow_run) mock_db.get_workflow_by_id = AsyncMock(return_value=workflow) mock_db.update_workflow_run = AsyncMock() @@ -255,3 +393,4 @@ async def test_smallwebrtc_run_reaching_telephony_websocket_closes_without_runni ) assert mock_db.update_workflow_run.await_count == 0 assert provider_lookup.await_count == 0 + mock_concurrency.unregister_active_call.assert_awaited_once_with(501) diff --git a/api/tests/test_webrtc_signaling_concurrency.py b/api/tests/test_webrtc_signaling_concurrency.py new file mode 100644 index 00000000..ae6203fc --- /dev/null +++ b/api/tests/test_webrtc_signaling_concurrency.py @@ -0,0 +1,117 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from api.routes.webrtc_signaling import SignalingManager +from api.services.call_concurrency import CallConcurrencyLimitError + + +class _FakeWebSocket: + def __init__(self): + self.send_json = AsyncMock() + + +class _FakePeerConnection: + def __init__(self): + self.renegotiate = AsyncMock() + + def get_answer(self): + return {"sdp": "v=0\r\n", "type": "answer", "pc_id": "pc-1"} + + +def _offer_payload(pc_id: str = "pc-1") -> dict: + return { + "pc_id": pc_id, + "sdp": "v=0\r\n", + "type": "offer", + } + + +@pytest.mark.asyncio +async def test_public_embed_offer_rejects_when_org_concurrency_limit_reached(): + manager = SignalingManager() + ws = _FakeWebSocket() + user = SimpleNamespace(id=7) + + with ( + patch("api.routes.webrtc_signaling.db_client") as mock_db, + patch( + "api.routes.webrtc_signaling.authorize_workflow_run_start", + new=AsyncMock( + return_value=SimpleNamespace(has_quota=True, error_message="") + ), + ), + patch("api.routes.webrtc_signaling.call_concurrency") as mock_concurrency, + ): + mock_db.get_workflow_organization_id = AsyncMock(return_value=11) + mock_concurrency.acquire_org_slot = AsyncMock( + side_effect=CallConcurrencyLimitError( + organization_id=11, + source="public_embed", + wait_time=0, + max_concurrent=2, + ) + ) + mock_concurrency.bind_workflow_run = AsyncMock() + + await manager._handle_offer( + ws, + _offer_payload(), + workflow_id=33, + workflow_run_id=501, + user=user, + connection_key="conn-1", + enforce_call_concurrency=True, + call_concurrency_source="public_embed", + ) + + ws.send_json.assert_awaited_once_with( + { + "type": "error", + "payload": { + "error_type": "concurrency_limit_exceeded", + "message": "Concurrent call limit reached", + }, + } + ) + mock_concurrency.bind_workflow_run.assert_not_called() + + +@pytest.mark.asyncio +async def test_public_embed_renegotiation_does_not_acquire_another_slot(): + manager = SignalingManager() + ws = _FakeWebSocket() + user = SimpleNamespace(id=7) + connection_key = "conn-1" + pc = _FakePeerConnection() + manager._peer_connections["pc-1"] = pc + manager._peer_connection_owners["pc-1"] = connection_key + + with ( + patch("api.routes.webrtc_signaling.db_client") as mock_db, + patch( + "api.routes.webrtc_signaling.authorize_workflow_run_start", + new=AsyncMock( + return_value=SimpleNamespace(has_quota=True, error_message="") + ), + ), + patch("api.routes.webrtc_signaling.call_concurrency") as mock_concurrency, + ): + mock_db.get_workflow_organization_id = AsyncMock(return_value=11) + mock_concurrency.acquire_org_slot = AsyncMock() + + await manager._handle_offer( + ws, + _offer_payload(), + workflow_id=33, + workflow_run_id=501, + user=user, + connection_key=connection_key, + enforce_call_concurrency=True, + call_concurrency_source="public_embed", + ) + + mock_concurrency.acquire_org_slot.assert_not_called() + pc.renegotiate.assert_awaited_once() + assert ws.send_json.await_args.args[0]["type"] == "answer" diff --git a/docs/developer/environment-variables.mdx b/docs/developer/environment-variables.mdx index 49984675..cf64b9d8 100644 --- a/docs/developer/environment-variables.mdx +++ b/docs/developer/environment-variables.mdx @@ -179,7 +179,7 @@ Controls concurrency for [Campaigns](/core-concepts/campaigns), Dograh's bulk ou | Variable | Default | Description | |---|---|---| -| `DEFAULT_ORG_CONCURRENCY_LIMIT` | `2` | Maximum concurrent outbound calls per organization | +| `DEFAULT_ORG_CONCURRENCY_LIMIT` | `10` | Maximum concurrent active calls per organization | --- diff --git a/ui/src/app/workflow/[workflowId]/run/[runId]/hooks/useWebSocketRTC.tsx b/ui/src/app/workflow/[workflowId]/run/[runId]/hooks/useWebSocketRTC.tsx index fb280f0c..2b17a9fe 100644 --- a/ui/src/app/workflow/[workflowId]/run/[runId]/hooks/useWebSocketRTC.tsx +++ b/ui/src/app/workflow/[workflowId]/run/[runId]/hooks/useWebSocketRTC.tsx @@ -393,8 +393,10 @@ export const useWebSocketRTC = ({ workflowId, workflowRunId, accessToken, initia // Stop the connection and surface the handled service error. cleanupConnection({ graceful: false, status: 'failed' }); } else { - // Log other errors as actual errors + const serverErrorMessage = message.payload?.message || 'Server error'; logger.error('Server error:', message.payload); + setPermissionError(serverErrorMessage); + cleanupConnection({ graceful: false, status: 'failed' }); } break; @@ -606,7 +608,7 @@ export const useWebSocketRTC = ({ workflowId, workflowRunId, accessToken, initia } }; }); - }, [getWebSocketUrl, cleanupConnection]); + }, [getWebSocketUrl, cleanupConnection, setPermissionError]); const negotiate = async () => { const pc = pcRef.current; @@ -661,6 +663,7 @@ export const useWebSocketRTC = ({ workflowId, workflowRunId, accessToken, initia setIsStarting(true); setConnectionActive(false); setIsCompleted(false); + setPermissionError(null); setConnectionStatus('connecting'); try {