117K+ emails and SMS, plus about 10K AI voice calls, in 6 hours on GCP Pub/Sub

The problem: human SDRs cannot scale

An inside sales team can only write so many personalized emails a day, and it is expensive: salaries, benefits and management overhead, plus inconsistent quality and limited operating hours.

The ask at Speculo.ai was to reach the client's lead lists every week, each lead with a message written for them, and to follow up with a voice call if they did not respond. With people doing the work, the unit economics fail at that volume.

Generating good copy with GPT-4o is straightforward. The hard part was infrastructure that could push 117K+ email and SMS sends and about 10K calls through in parallel without losing messages or hitting rate limits, and with full observability.

Architecture overview

The system has four layers, decoupled by GCP Pub/Sub topics. Each layer publishes to the next. Workers in each layer are stateless Cloud Run containers that scale horizontally based on subscription backlog.

The four layers are:

Why GCP Pub/Sub and not a task queue?

I evaluated Celery with Redis, BullMQ, and Google Cloud Tasks before settling on Pub/Sub. The decision came down to three requirements:

1. Fan-out publishing. Some leads need email only. Some get email and a follow-up call. Some are filtered and dropped. With Pub/Sub, a single enrichment worker can publish one message and multiple downstream subscribers decide independently whether to act. Cloud Tasks would have required explicit routing logic in the publisher.

2. At-least-once delivery, made safe with idempotency. Pub/Sub delivers every message at least once. If a worker crashes or misses its ack deadline, the message is redelivered to another instance, so nothing is lost, but the same message can arrive twice. At 117K+ messages, a duplicate email or SMS is a reputational disaster, so every send carried an idempotency key (the lead ID plus the channel). The dispatch worker records that key before sending and skips any message whose key is already there, which makes redelivery harmless.

3. Visible backlog for scaling and alerts. Each worker pool pulls from its own subscription, and the subscription backlog is a first-class metric in Cloud Monitoring. That made it simple to size the Cloud Run workers per stage and to alert when a stage fell behind.

The four Pub/Sub topics

Topic 1: leads-raw

Every lead from the client's lead lists is published here after a lightweight Cloud Function validates the schema, deduplicates against a Firestore set of seen lead IDs and computes an initial lead score. Malformed records are sent directly to a dead letter topic.

import json
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
TOPIC = "projects/speculo-prod/topics/leads-raw"

def ingest_lead(lead: dict) -> str:
    if not is_valid(lead):
        publish_dlq(lead, reason="schema_invalid")
        return "rejected"

    lead["lead_score"] = score_lead(lead)
    lead["lead_id"]    = deduplicate_id(lead)   # stable hash of the contact details

    future = publisher.publish(
        TOPIC,
        data=json.dumps(lead).encode("utf-8"),
        source=lead.get("source", "unknown"),   # message attributes for filtering
    )
    return future.result()  # blocks only until the broker acks

Topic 2: leads-enriched

Enrichment workers subscribe to leads-raw, check the contact details, summarize the lead record and publish the enriched payload here. The ack deadline is set to 120 seconds to accommodate slow API responses. If enrichment fails after 3 retries, the message routes to leads-raw-dlq with a failure reason tag.

import asyncio
from google.cloud import pubsub_v1
from agents import summarize_lead, verify_contact

subscriber = pubsub_v1.SubscriberClient()
PULL_SUB   = "projects/speculo-prod/subscriptions/enrichment-sub"
PUSH_TOPIC = "projects/speculo-prod/topics/leads-enriched"

async def process(message: pubsub_v1.types.ReceivedMessage):
    lead = json.loads(message.message.data)
    try:
        lead_ctx, contact = await asyncio.gather(
            summarize_lead(lead),     # Gemini summary of the lead record
            verify_contact(lead),     # MX check and contact validation
        )
        if not contact["deliverable"]:
            message.nack()   # requeue for DLQ after max attempts
            return

        enriched = {**lead, "lead_ctx": lead_ctx, "contact": contact}
        publisher.publish(PUSH_TOPIC, json.dumps(enriched).encode())
        message.ack()
    except Exception as e:
        log_error(e, lead["lead_id"])
        message.nack()   # retry with exponential backoff

Topic 3: messages-ready

The personalization layer is where the LLM calls happen. Each worker receives an enriched lead and generates three things: an email subject line, an HTML email body (~120 words, written for that lead), and a 60-second call script for the voice agent. All three are generated in a single LLM call using structured output (JSON mode) to avoid parsing failures.

async def personalize(enriched: dict) -> dict:
    prompt = build_prompt(
        sender     = enriched["agent_profile"],   # the client's agent / brokerage
        lead       = enriched["lead_ctx"],
        contact    = enriched["contact"],
    )
    response = await openai.chat.completions.create(
        model           = "gpt-4o",
        messages        = [{"role": "user", "content": prompt}],
        response_format = {"type": "json_object"},
        temperature     = 0.7,
        max_tokens      = 800,
    )
    result = json.loads(response.choices[0].message.content)

    # Hard quality gate: discard and requeue if score < 8
    if result.get("review_score", 0) < 8:
        return requeue_for_repersonalization(enriched)

    return {
        **enriched,
        "subject"     : humanize(result["subject"]),   # strip AI tells
        "body_html"   : humanize(result["body_html"]),
        "call_script" : result["call_script"],
        "channel"     : determine_channel(enriched),   # email | sms | voice | both
    }

The hardest problem: rate limiting at scale

With 50 enrichment workers and 30 personalization workers all hitting the same LLM APIs simultaneously, you will hit your rate limits in minutes. This was the most painful part of the build. The rule we followed was simple: stay under the paid quota we had with each provider, and make workers wait instead of fail.

A token bucket per API key

Each API key has a shared limiter in Cloud Memorystore (Redis), sized just under that account's paid requests-per-minute quota. Every worker takes a token before making an API call. The version below is the simplest form of a token bucket: a one-minute window that refills all at once. If the bucket is empty, the worker sleeps and retries with exponential backoff. It keeps the Pub/Sub message (within its ack deadline) instead of nacking it, so a busy minute slows the pipeline down rather than pushing messages toward the dead letter queue.

import asyncio
import os
import time
import redis

r = redis.Redis.from_url(os.environ["MEMORYSTORE_URL"])

def acquire_token(api_key: str, rpm_limit: int) -> bool:
    """
    Returns True if token acquired, False if this minute's quota is used up.
    Uses Redis atomic increment with a TTL-based one-minute window.
    rpm_limit is set just below the provider's paid quota for this key.
    """
    bucket_key = f"ratelimit:{api_key}:{int(time.time() // 60)}"
    count = r.incr(bucket_key)
    if count == 1:
        r.expire(bucket_key, 65)   # slightly over 60s to avoid edge cases
    return count <= rpm_limit

async def call_with_backoff(fn, api_key: str, rpm: int, max_attempts=5):
    for attempt in range(max_attempts):
        if acquire_token(api_key, rpm):
            return await fn()
        wait = min(4 ** attempt, 30)   # 1s, 4s, 16s, 30s cap
        await asyncio.sleep(wait)
    raise RateLimitExhausted(f"Key {api_key[:8]}... exhausted after {max_attempts} attempts")

Treat the provider's own 429 responses the same way: back off and retry. The limiter keeps you under the quota, and the backoff covers the moments it gets it wrong.

Email and SMS dispatch: idempotency keys against duplicate sends

The dispatch layer had one non-negotiable requirement: no lead gets the same email or SMS twice, even if a worker crashes mid-send and Pub/Sub redelivers the message. Because delivery is at-least-once, the guard has to live in the worker. Each send has an idempotency key (lead ID plus channel), and the worker claims that key in a Firestore transaction before it sends anything:

from google.cloud import firestore

db = firestore.Client()

async def dispatch_message(message: dict) -> bool:
    send_key = f"{message['lead_id']}:{message['channel']}"   # idempotency key
    send_ref = db.collection("sends").document(send_key)

    # Step 1: claim the key (atomic transaction)
    @firestore.transactional
    def claim(txn):
        snap = send_ref.get(transaction=txn)
        if snap.exists and snap.get("status") != "failed":
            return False   # already sent or in flight: idempotent skip
        txn.set(send_ref, {"status": "pending", "ts": firestore.SERVER_TIMESTAMP})
        return True

    txn = db.transaction()
    if not claim(txn):
        return False   # duplicate delivery: ack and move on

    # Step 2: actually send
    try:
        await send(message)   # email or SMS, by channel
        send_ref.update({"status": "sent", "sent_at": firestore.SERVER_TIMESTAMP})
        stream_to_bigquery(message, event=f"{message['channel']}_sent")
        return True
    except Exception as e:
        send_ref.update({"status": "failed", "error": str(e)})
        raise   # Pub/Sub redelivers; "failed" keys can be claimed again

At this volume, sending domains have to be warmed up before you use them. We used three warmed sending domains with SPF, DKIM and DMARC records, and rotated messages across them to avoid triggering spam filters. The workers enforced per-domain hourly sending caps to protect deliverability.

Voice dispatch: about 10K AI calls through Twilio and ElevenLabs

Voice calls used the same Pub/Sub pattern as email, with a separate subscription on messages-ready filtered to leads with channel=voice or channel=both.

Calls ran on Twilio Programmable Voice with an ElevenLabs voice; a LangGraph state machine handled the qualification script and passed qualified leads to the client's human agents. Call transcripts were written to BigQuery when each call ended.

Observability: BigQuery as the source of truth

Every event that mattered (lead ingested, enrichment succeeded, email sent, call completed, reply received) streamed to a BigQuery dataset in real time through the Storage Write API. That gave us one queryable audit log for the whole pipeline.

# events table, partitioned by DATE(event_ts)
CREATE TABLE speculo.pipeline_events (
  lead_id       STRING NOT NULL,
  event_type    STRING NOT NULL,   -- ingest | enrich | personalize | email_sent | sms_sent | call_started | reply
  event_ts      TIMESTAMP,
  contact_email STRING,
  channel       STRING,
  metadata      JSON,              -- flexible per-event payload
  worker_id     STRING,            -- Cloud Run instance ID for debugging
)
PARTITION BY DATE(event_ts)
CLUSTER BY event_type, channel;

-- Daily funnel query
SELECT
  event_type,
  COUNT(*) AS count,
  COUNT(*) / LAG(COUNT(*)) OVER (ORDER BY stage_order) AS conversion_rate
FROM speculo.pipeline_events
WHERE DATE(event_ts) = CURRENT_DATE()
GROUP BY event_type, stage_order
ORDER BY stage_order;

Cloud Monitoring dashboards tracked Pub/Sub backlog per topic, Cloud Run instance count, and p95 processing latency per layer. Alerting fired if any topic backlog exceeded 5,000 messages for more than 10 minutes, a sign that a layer was unhealthy.

Results

One production run sent 117K+ emails and SMS and made about 10K AI voice calls in 6 hours, and the idempotency keys prevented duplicate sends. The qualified leads went to the client's human agents, and the work contributed to $2M+ in annual revenue.

What I would do differently

Separate the quality gate into its own service. In the current design, the personalization worker scores the output and requeues failures. Failed messages re-enter the personalization queue and count against our LLM quota. A dedicated scoring worker with no LLM (rules and fast heuristics) would catch the obvious failures before spending tokens on them.

Implement proper send-time scheduling.We sent at a constant rate through the run, with no regard for the recipient's local time. A send-time layer that holds each message until a reasonable hour in the recipient's time zone would fix that.