# How the Dograh Campaign Dispatcher Manages Outbound Calling at Scale

> Discover how the Dograh campaign dispatcher handles large scale outbound calling using ARQ workers, Redis rate limiting, circuit breakers, and Pipecat. Learn to protect APIs and scale effectively.

- Repository: [Dograh/dograh](https://github.com/dograh-hq/dograh)
- Tags: how-to-guide
- Published: 2026-05-18

---

**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`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/fefdd1835b7d_retry_outbound_calls_for_campaigns.py).

## Implementation Examples

### Starting the Dispatcher Worker

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

```bash

# 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:

```python
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:

```python
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:

```typescript
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`](https://github.com/dograh-hq/dograh/blob/main/api/services/campaign/campaign_call_dispatcher.py) — Core dispatcher coordinating rate limiting, circuit breaking, and Pipecat execution.
- [`api/services/campaign/campaign_orchestrator.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/campaign/campaign_orchestrator.py) — Creates call rows and queues ARQ jobs.
- [`api/services/campaign/rate_limiter.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/campaign/rate_limiter.py) — Redis-backed token-bucket implementation.
- [`api/services/campaign/circuit_breaker.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/campaign/circuit_breaker.py) — Failure threshold monitoring and tripping logic.
- [`api/services/campaign/campaign_event_publisher.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/campaign/campaign_event_publisher.py) — Typed event emission for UI and analytics.
- [`api/tasks/campaign_tasks.py`](https://github.com/dograh-hq/dograh/blob/main/api/tasks/campaign_tasks.py) — ARQ task definitions invoking the dispatcher.
- [`api/tests/test_campaign_call_dispatcher.py`](https://github.com/dograh-hq/dograh/blob/main/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`](https://github.com/dograh-hq/dograh/blob/main/campaign_tasks.py). The system uses columns added by Alembic migration [`fefdd1835b7d_retry_outbound_calls_for_campaigns.py`](https://github.com/dograh-hq/dograh/blob/main/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.