How the Dograh Campaign Dispatcher Manages Outbound Calling at Scale

The Dograh campaign dispatcher combines ARQ distributed workers, Redis-backed token-bucket rate limiting, circuit breakers, and Pipecat telephony integration to execute thousands of concurrent outbound calls while protecting downstream APIs from overload.

The dograh-hq/dograh open-source platform provides a robust campaign dispatcher designed to manage outbound calling at scale. By leveraging asynchronous task queues and stateless worker architecture, the system distributes high-volume calling workloads across multiple processes while maintaining strict rate limits and real-time observability.

Core Architecture Components

Task Orchestration with CampaignOrchestrator

In api/services/campaign/campaign_orchestrator.py, the CampaignOrchestrator class creates call tasks by writing rows to the campaign_calls table for each contact. It then schedules background ARQ jobs that point the dispatcher at pending rows.

Distributed Work Queue via ARQ

The dispatcher runs as an ARQ worker defined in api/tasks/campaign_tasks.py. Each worker executes run_campaign_call, pulling batches of pending call rows, marking them in-flight, and handing them to the telephony layer. Because ARQ uses Redis as its backing store, horizontal scaling requires only adding more worker processes.

Token-Bucket Rate Limiting

Before placing any call, the dispatcher invokes RateLimiter.acquire from api/services/campaign/rate_limiter.py. This Redis-backed token-bucket implementation enforces per-campaign limits (e.g., 10 calls per second) globally across all workers, ensuring provider API limits are respected regardless of worker count.

Circuit Breaker Pattern

The CircuitBreaker class in api/services/campaign/circuit_breaker.py monitors downstream health. When Twilio or Pipecat rejections exceed configurable thresholds, the circuit trips and CircuitBreaker.allow() returns false, causing the dispatcher to pause new calls and re-queue tasks until the downstream service recovers.

Event Publishing and Observability

Every state change flows through CampaignEventPublisher in api/services/campaign/campaign_event_publisher.py. Typed events (queued, dialing, answered, failed) propagate via the internal Pub/Sub system, keeping UIs synchronized through WebSocket connections while enabling audit trails and metrics collection.

End-to-End Call Flow

1. Contact enrichment pulls data from CSV or Google Sheets into the campaign database.
2. Task creation writes rows to campaign_calls and schedules ARQ jobs via CampaignOrchestrator.schedule_calls.
3. Worker pickup executes CampaignCallDispatcher.dispatch via campaign_tasks.run_campaign_call.
4. Rate limiting checks RateLimiter.acquire for token availability; if exhausted, the task re-queues after a back-off delay.
5. Circuit breaker validates downstream health via CircuitBreaker.allow before proceeding.
6. Telephony execution hands CallContext to the Pipecat submodule for asynchronous call placement.
7. Event emission publishes state transitions via CampaignEventPublisher.publish for real-time UI updates.
8. Result handling updates the database and schedules retries with exponential back-off, tracked via columns added in Alembic migration fefdd1835b7d_retry_outbound_calls_for_campaigns.py.

Implementation Examples

Starting the Dispatcher Worker

Run the ARQ worker process to begin consuming campaign call tasks:

# Activate the virtualenv and load env vars

source venv/bin/activate
set -a && source api/.env.test && set +a

# Run the ARQ worker that processes campaign call tasks

python -m arq worker api.tasks.campaign_tasks

Enqueueing Calls for Contacts

Schedule outbound calls by invoking the orchestrator:

from api.services.campaign.campaign_orchestrator import CampaignOrchestrator

# Assume `campaign_id` and a list of contacts already exist

orchestrator = CampaignOrchestrator(db_session)
await orchestrator.schedule_calls(
    campaign_id=campaign_id,
    contacts=[{"phone": "+15551234567", "name": "Alice"}],
)

Manual Rate Limiter Usage

Check rate limits before dispatching:

from api.services.campaign.rate_limiter import RateLimiter

limiter = RateLimiter(redis_client, key="campaign:42:rate")
if await limiter.acquire():
    # safe to place a call

    await dispatcher.dispatch(call_row)
else:
    # re-queue after a short back-off

    await scheduler.enqueue_in(5, dispatch, call_row)

Listening to Campaign Events in the UI

Connect to the real-time event stream via WebSocket:

import { createEventSource } from "@/utils/eventSource";

const events = createEventSource(
  `/api/campaigns/${campaignId}/events`
);

events.addEventListener("call_started", (e) => {
  const data = JSON.parse(e.data);
  console.log(`Call started for ${data.phone}`);
});

Key Source Files

• api/services/campaign/campaign_call_dispatcher.py — Core dispatcher coordinating rate limiting, circuit breaking, and Pipecat execution.
• api/services/campaign/campaign_orchestrator.py — Creates call rows and queues ARQ jobs.
• api/services/campaign/rate_limiter.py — Redis-backed token-bucket implementation.
• api/services/campaign/circuit_breaker.py — Failure threshold monitoring and tripping logic.
• api/services/campaign/campaign_event_publisher.py — Typed event emission for UI and analytics.
• api/tasks/campaign_tasks.py — ARQ task definitions invoking the dispatcher.
• api/tests/test_campaign_call_dispatcher.py — Load and failure condition verification.
• pipecat/ — Telephony abstraction submodule handling actual call streams.

Summary

• Stateless architecture allows linear scaling by adding ARQ workers without code changes.
• Redis-backed rate limiting provides global token-bucket throttling per campaign across all workers.
• Circuit breakers prevent cascading failures when telephony providers degrade or timeout.
• Pipecat integration enables provider-agnostic call execution (Twilio, SignalWire, etc.).
• Event-driven design ensures real-time UI synchronization and comprehensive observability.

Frequently Asked Questions

How does the dispatcher handle rate limiting across multiple workers?

The RateLimiter class uses Redis to maintain a token bucket shared across all worker processes. When acquire() is called, it atomically decrements tokens in Redis, ensuring consistent global limits regardless of worker count or distribution across machines.

What happens when a downstream telephony provider fails?

The CircuitBreaker monitors failure thresholds. After configurable consecutive failures, the circuit trips and CircuitBreaker.allow() returns false, causing the dispatcher to pause new calls and re-queue tasks with exponential back-off until the downstream service recovers.

How does the system retry failed calls?

Failed calls trigger retry logic in campaign_tasks.py. The system uses columns added by Alembic migration fefdd1835b7d_retry_outbound_calls_for_campaigns.py to track retry counts and next attempt timestamps, ensuring transient failures don't exhaust resources while maintaining delivery guarantees.

Can the dispatcher scale horizontally without code changes?

Yes. Because the architecture is stateless and uses Redis for coordination and the database for persistence, adding more ARQ worker processes or scaling the Redis cluster increases capacity linearly without modifying application code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}