---
title: "Backpressure in Elixir: sending millions of WhatsApp messages without falling over"
description: "Designing a bulk WhatsApp send engine in Elixir: demand-driven Broadway pipelines, per-number token buckets, retries with jitter and batched status tracking."
author: Federico Meini
date: 2026-08-27
tags: [elixir, whatsapp, backpressure, broadway, oban]
language: en
url: https://fedme.dev/blog/backpressure-bulk-whatsapp-sends-elixir
---

# Backpressure in Elixir: sending millions of WhatsApp messages without falling over

To send millions of WhatsApp messages without falling over, the system has to pull work at the rate the slowest part can absorb instead of pushing everything at once. In practice that means a demand-driven pipeline that claims recipients from Postgres in small batches, a token bucket per business phone number that paces calls to the Cloud API, throughput errors that slow the whole number down instead of triggering a retry storm, and delivery statuses written back in batches rather than one row at a time. At Turn.io I built the high-throughput bulk send engine used for national public-health campaigns reaching 14.7 million people, covering rate limiting, backpressure, delivery tracking and retries. This post is the design I'd use for that problem today, with Elixir sketches for each piece.

## TL;DR

- Keep **durable state per recipient in Postgres** and treat in-memory pipeline stages as disposable.
- Use **GenStage/Broadway** so the API's pace drives how fast you read from the database, not the other way round.
- Put a **token bucket per business phone number** in front of the API. Throughput is a per-number limit, so the limiter has to be too.
- Classify errors: **throughput errors pause the number**, per-recipient limits reschedule that recipient, permanent errors fail fast, and ambiguous ones (timeouts) are reconciled, not blindly retried.
- Retry with **exponential backoff and full jitter**.
- Status webhooks arrive at several times your send rate and **out of order**: apply them monotonically and in batches.
- Instrument everything with `:telemetry` and export to Prometheus. You can't tune what you can't see.

## The constraints you're designing against

Three limits shape the whole design.

**Cloud API throughput.** Throughput is per business phone number. By default the Cloud API allows up to 80 messages per second per number, and eligible numbers can be upgraded to more (up to 1,000 at the time of writing; check Meta's current documentation for your numbers). Exceed it and you get throughput errors (error code 130429). There is also a per-recipient pair rate limit (131056) if you send too many messages to the same user in a short window, and messaging limits on how many unique users you can reach with business-initiated messages in a rolling 24-hour period, which depend on your account's standing.

**Your own database.** Every send is at least one read and one write. Every sent message then generates status webhooks (`sent`, `delivered`, `read` or `failed`), so status writes run at a multiple of the send rate. It's easy to size everything around the API limit and forget that the database has to keep up with all of this.

**Shared capacity.** Your business number is probably also handling live conversations. A campaign that uses 100% of a number's throughput makes the service unresponsive to people replying to it. Budget headroom on purpose.

## Architecture

```text
campaign_recipients (Postgres, source of truth)
        │  claim N rows (FOR UPDATE SKIP LOCKED), on demand
        ▼
RecipientProducer ──► processors (acquire token ► call Cloud API)
                                    │
                                    ▼
                         batcher: write results in bulk
```

Nothing is pushed. Processors ask for more work when they're free, the producer only claims as many rows as there is demand for, and the token bucket blocks processors when the number is at capacity. If the API slows down, processors take longer, demand drops, and the producer stops reading from the database. That's backpressure: every stage slows down together instead of queues growing somewhere in the middle.

## Oban jobs or in-memory stages?

A natural first design is one Oban job per message. It's durable, retries are built in, and it works up to a point. At millions of messages per campaign, though, you're inserting millions of job rows, updating each one several times as it moves through states, and pruning them afterwards. The jobs table becomes the hottest, most bloated table in your database, and throughput control is spread across queue configuration instead of living in one place.

What works better is splitting responsibilities:

- **Postgres rows are the durable state**: one row per recipient with a status, an attempt count and a `next_attempt_at`.
- **Oban handles orchestration**: starting a campaign, scheduling it, periodically reclaiming rows stuck in `sending` after a crash, marking the campaign complete. A handful of jobs per campaign, not one per message.
- **GenStage/Broadway does the sending**: fast, in memory, disposable. If a node dies, the pipeline restarts and picks up from the table.

## Claiming work without contention

The claim is a single statement. `FOR UPDATE SKIP LOCKED` lets several producers (or nodes) claim concurrently without blocking each other or taking the same row:

```sql
UPDATE campaign_recipients
SET status = 'sending', claimed_at = now()
WHERE id IN (
  SELECT id FROM campaign_recipients
  WHERE campaign_id = $1
    AND status = 'pending'
    AND next_attempt_at <= now()
  ORDER BY next_attempt_at
  LIMIT $2
  FOR UPDATE SKIP LOCKED
)
RETURNING id, phone, template_params, attempt;
```

Back it with a partial index so the claim stays cheap as the table fills with finished rows:

```sql
CREATE INDEX campaign_recipients_pending_index
  ON campaign_recipients (campaign_id, next_attempt_at)
  WHERE status = 'pending';
```

## A demand-driven producer

Broadway takes any GenStage producer. This one only touches the database when downstream stages have asked for messages, and polls when it runs dry:

```elixir
defmodule Bulk.RecipientProducer do
  use GenStage
  @behaviour Broadway.Producer

  alias Broadway.Message

  @max_batch 500
  @poll_interval 1_000

  @impl true
  def init(opts) do
    state = %{campaign_id: Keyword.fetch!(opts, :campaign_id), demand: 0, poll_scheduled?: false}
    {:producer, state}
  end

  @impl true
  def handle_demand(incoming, state), do: claim(%{state | demand: state.demand + incoming})

  @impl true
  def handle_info(:poll, state), do: claim(%{state | poll_scheduled?: false})

  defp claim(%{demand: 0} = state), do: {:noreply, [], state}

  defp claim(state) do
    limit = min(state.demand, @max_batch)
    recipients = Bulk.Recipients.claim(state.campaign_id, limit)

    messages =
      Enum.map(recipients, fn r ->
        %Message{data: r, acknowledger: Broadway.NoopAcknowledger.init()}
      end)

    state = %{state | demand: state.demand - length(messages)}
    {:noreply, messages, schedule_poll(state, length(messages) == limit)}
  end

  # A full batch means more rows are probably waiting: poll again straight away.
  # Otherwise we've caught up (or what's left is in backoff): check again later.
  defp schedule_poll(%{demand: 0} = state, _full?), do: state
  defp schedule_poll(%{poll_scheduled?: true} = state, _full?), do: state

  defp schedule_poll(state, full?) do
    Process.send_after(self(), :poll, if(full?, do: 0, else: @poll_interval))
    %{state | poll_scheduled?: true}
  end
end
```

Acknowledgement is a no-op because the database row, not the message, is the unit of durability. Results are recorded by the batcher.

## A token bucket per phone number

Broadway has a built-in `rate_limiting` option on the producer, and if one pipeline maps to one phone number it's a fine start. In practice several campaigns and the conversational traffic share a number, so I want one limiter per number that everything sending from it goes through. A small GenServer does the job:

```elixir
defmodule Bulk.TokenBucket do
  use GenServer

  def start_link(opts) do
    id = Keyword.fetch!(opts, :phone_number_id)
    GenServer.start_link(__MODULE__, opts, name: via(id))
  end

  @doc "Blocks the caller until this number has capacity for one more send."
  def acquire(phone_number_id) do
    case GenServer.call(via(phone_number_id), :take) do
      :ok ->
        :ok

      {:wait, ms} ->
        Process.sleep(ms)
        acquire(phone_number_id)
    end
  end

  @doc "Stops all sends from this number for `ms`, e.g. after a throughput error."
  def pause(phone_number_id, ms), do: GenServer.cast(via(phone_number_id), {:pause, ms})

  defp via(id), do: {:via, Registry, {Bulk.Registry, {:bucket, id}}}

  @impl true
  def init(opts) do
    rate = Keyword.fetch!(opts, :rate)
    {:ok, %{rate: rate, tokens: rate * 1.0, updated_at: now()}}
  end

  @impl true
  def handle_call(:take, _from, state) do
    now = now()
    state = refill(state, now)

    cond do
      now < state.updated_at -> {:reply, {:wait, state.updated_at - now}, state}
      state.tokens >= 1 -> {:reply, :ok, %{state | tokens: state.tokens - 1}}
      true -> {:reply, {:wait, ceil((1 - state.tokens) * 1000 / state.rate)}, state}
    end
  end

  @impl true
  def handle_cast({:pause, ms}, state) do
    resume_at = max(state.updated_at, now() + ms)
    {:noreply, %{state | tokens: 0.0, updated_at: resume_at}}
  end

  # While paused, updated_at is in the future and no tokens accrue.
  defp refill(state, now) when now <= state.updated_at, do: state

  defp refill(state, now) do
    tokens = min(state.rate * 1.0, state.tokens + (now - state.updated_at) * state.rate / 1000)
    %{state | tokens: tokens, updated_at: now}
  end

  defp now, do: System.monotonic_time(:millisecond)
end
```

Set `rate` below the number's real limit to leave room for live conversations. Waiting happens in the caller, so the bucket process never blocks. One GenServer call per send is nothing at these rates.

Two things to get right around it. First, the bucket must have a **single owner per number** across the cluster. `Registry` is node-local, so either run each number's sending on one node (a global registry, or routing numbers to nodes) or move the limiter somewhere shared. Two nodes with their own buckets at 80 per second each will cheerfully send 160. Second, **concurrency has to cover latency**. By Little's law, in-flight requests equal rate times latency: to sustain, say, 80 messages per second with 250 ms API latency you need at least 20 concurrent processors, plus margin for slow responses.

## The pipeline

```elixir
defmodule Bulk.Pipeline do
  use Broadway

  alias Broadway.Message

  def start_link(campaign) do
    Broadway.start_link(__MODULE__,
      name: {:via, Registry, {Bulk.Registry, {:pipeline, campaign.id}}},
      producer: [module: {Bulk.RecipientProducer, campaign_id: campaign.id}, concurrency: 1],
      processors: [default: [concurrency: 40, max_demand: 5]],
      batchers: [default: [concurrency: 1, batch_size: 500, batch_timeout: 1_000]],
      context: %{phone_number_id: campaign.phone_number_id}
    )
  end

  @impl true
  def process_name({:via, Registry, {registry, key}}, base_name) do
    {:via, Registry, {registry, {key, base_name}}}
  end

  @impl true
  def handle_message(_processor, %Message{data: recipient} = message, ctx) do
    :ok = Bulk.TokenBucket.acquire(ctx.phone_number_id)

    result =
      :telemetry.span([:bulk, :send], %{phone_number_id: ctx.phone_number_id}, fn ->
        result = WhatsApp.send_template(ctx.phone_number_id, recipient)
        {result, %{phone_number_id: ctx.phone_number_id, ok?: match?({:ok, _}, result)}}
      end)

    Message.put_data(message, {recipient, classify(result, recipient, ctx)})
  end

  @impl true
  def handle_batch(:default, messages, _batch_info, _ctx) do
    messages |> Enum.map(& &1.data) |> Bulk.Recipients.record_results()
    messages
  end
end
```

Small `max_demand` on processors matters. With large demand, each processor buffers many messages it can't send yet because it's waiting on the bucket, and those rows sit in `sending` doing nothing.

## Errors, retries and backoff with jitter

Not all errors mean the same thing, and treating them the same is how retry storms start:

```elixir
# Accepted by the API: record the WhatsApp message id.
defp classify({:ok, wamid}, _recipient, _ctx), do: {:sent, wamid}

# Throughput reached for this number: slow the whole number down, retry later.
defp classify({:error, %{code: 130429}}, r, ctx) do
  Bulk.TokenBucket.pause(ctx.phone_number_id, 1_000)
  {:retry, Bulk.Backoff.delay_ms(r.attempt)}
end

# Too many messages to this one user: reschedule only this recipient.
defp classify({:error, %{code: 131056}}, r, _ctx), do: {:retry, Bulk.Backoff.delay_ms(r.attempt)}

# We don't know whether the message went out: reconcile, don't resend blindly.
defp classify({:error, :timeout}, _r, _ctx), do: :unknown

# Everything else is treated as permanent for this recipient.
defp classify({:error, error}, _r, _ctx), do: {:failed, error}
```

And the backoff, using "full jitter": a random delay between zero and an exponentially growing ceiling. Without jitter, every recipient that failed in the same second retries in the same second, and you hit the limit again in lockstep.

```elixir
defmodule Bulk.Backoff do
  @base_ms 1_000
  @cap_ms 5 * 60_000

  def delay_ms(attempt) do
    :rand.uniform(min(@cap_ms, @base_ms * Integer.pow(2, attempt)))
  end
end
```

Retries go back through the table: set the row to `pending`, bump `attempt`, set `next_attempt_at`. After a maximum number of attempts, mark it failed. Because retries are rows and not sleeping processes, a retry backlog costs nothing in memory and survives deploys.

## Idempotency and the "did it send?" problem

The dangerous case is a request that times out. Meta may have accepted it, or not. Assume the API won't deduplicate a resend for you, so you have to decide which is worse for this campaign: a duplicate message or a missed one. For health reminders, a duplicate is usually the lesser evil. For anything that looks like a payment prompt, it isn't.

Either way, make it reconcilable. Mark the row `sending` before calling the API, and pass your recipient id in `biz_opaque_callback_data` on the send request. The Cloud API echoes that field back in status webhooks, so even if you never stored the WhatsApp message id, a later `sent` or `delivered` status tells you the message did go out. A periodic Oban job then resolves rows stuck in `sending` or `unknown`: anything with a matching status webhook is sent, anything still silent after a generous window gets the campaign's chosen policy.

## Tracking delivery status in batches

Status webhooks arrive at a multiple of your send rate, and not necessarily in order: a `read` can arrive before its `delivered`. Two rules: never move a message backwards, and never write one row per webhook.

Rank the statuses and only apply an update that moves forward:

```sql
CREATE FUNCTION message_status_rank(text) RETURNS int
LANGUAGE sql IMMUTABLE AS $$
  SELECT CASE $1
    WHEN 'sending' THEN 0 WHEN 'sent' THEN 1 WHEN 'failed' THEN 2
    WHEN 'delivered' THEN 3 WHEN 'read' THEN 4
  END
$$;

UPDATE campaign_recipients r
SET status = s.status, status_at = s.at
FROM unnest($1::text[], $2::text[], $3::timestamptz[]) AS s(wamid, status, at)
WHERE r.wamid = s.wamid
  AND message_status_rank(s.status) > message_status_rank(r.status);
```

Buffer incoming statuses for up to a second or a few hundred entries, reduce them in Elixir to the highest-ranked status per message id (if the same id appears twice in one `UPDATE ... FROM`, Postgres applies only one of the matching rows, and you don't get to choose which), then run one statement for the whole batch. The webhook receiver itself should do nothing but verify, persist and acknowledge; I've written up [how I build that part](https://fedme.dev/blog/whatsapp-cloud-api-webhooks-at-scale).

## Observability

Broadway emits `:telemetry` events for its stages, and the `:telemetry.span/3` around the API call gives you start/stop/exception events with durations. Export them with `telemetry_metrics_prometheus` (or PromEx) and build a dashboard with at least:

- sends per second per phone number, against the configured rate
- time spent waiting in `TokenBucket.acquire` (if it's always high, you're bucket-bound; if it's zero and throughput is low, the bottleneck is elsewhere)
- API latency percentiles and error counts by error code
- claim query duration and number of `pending` rows per campaign
- time from `sent` to `delivered`, which tells you about the recipients' side, not yours
- rows stuck in `sending` for longer than a few minutes

Alert on throughput errors, not just on failures. A steady trickle of 130429s means your configured rate is higher than what Meta is actually giving you.

## Mistakes worth avoiding

- **Reading the whole audience into memory** at campaign start. Claim in small batches.
- **Retrying inside the processor with `Process.sleep`.** It holds a processor and a row hostage. Put retries back in the table.
- **Letting several nodes share a number without coordinating.** Per-number limits need a per-number owner.
- **Forgetting the conversational traffic.** When a million people receive a message, some of them reply. Your inbound path needs capacity at exactly the moment the campaign peaks.

## Need a bulk sender that holds up?

A bulk messaging engine is a rate-limited, stateful, distributed system disguised as a loop over a list. If you're building one for WhatsApp, or your current one struggles at campaign peaks, [I help teams design and build real-time systems like this](https://fedme.dev/services/realtime-systems).