Distributed Cron — Mass Scheduled Email — a worked solution

What each component is for

Trigger Source

Two distinct callers share this entry point: (a) the campaign-owner UI / internal service POSTing a campaign definition (template_id, recipient_list_ref, scheduled_at, priority) — happens dozens to hundreds of times per day; (b) the scheduler tier ITSELF self-firing at scheduled_at — happens internally and does not flow through the public gateway. We model both as 'clients' of the system because both originate writes that mutate state; the scheduler's self-fires are drawn as a separate edge from the scheduler node directly into the Kafka spine.

Why it exists. Considered modeling the cron tick as an implicit internal-only event (not a node). Rejected because the load-bearing semantic of this whole problem is 'a single event becomes 50M', and hiding the originator obscures where the at-least-once-then-dedupe contract begins. Surfacing the trigger as a node makes the Borgcron pattern visible: the trigger LANDS in a durable log before fan-out, and that log is what survives scheduler failover.

When it fails. Naïve callers regenerate campaign_id on retry → ghost campaigns + double-fires. Detection: distinct_campaign_ids_per_logical_submit_p95 > 1 alerts SDK eng. Mitigation: official SDKs ship the canonical retry helper; broken callers caught at gateway via HMAC fingerprint mismatch (Stripe pattern). The cron-tick caller (internal scheduler self-fire) cannot mis-key — its fire_id = (campaign_id, scheduled_at) is deterministic.

API Gateway · WAF + Auth + Rate-limitEnvoy 1.30 + Cloudflare WAF + per-tenant token-bucket

L7 ingress for the campaign control plane and the SES webhook intake. Terminates TLS, runs the WAF (OWASP top-10 + bulk-email-spam detector), enforces token-bucket rate limits per tenant BEFORE forwarding so a misbehaving caller cannot poison the idempotency store, validates the bearer token against the auth service (cached 60 s), copies tenant_id + Idempotency-Key headers verbatim onto the upstream request, and HMAC-signs forwarded headers so the acceptor trusts tenant_id without re-validating. SES bounce webhooks land on a separate path (/webhooks/ses) that verifies SES SNS signatures before forwarding to webhook-rx.

Why it exists. Considered putting auth + rate limit inline in the campaign-api. Rejected because (1) a 5xx in the acceptor would skip the rate limiter and let a retry storm reach Cassandra LWT — at $0.000005/LWT write, even a small storm wastes budget AND triggers Paxos contention; (2) per-tenant rate-limit state lives in Redis which we don't want to ship to every campaign-api replica. Stripe's 2019-07-10 retro is the precedent: rate limits in the wrong layer cascade across product surfaces.

When it fails. Edge pool exhaustion masks as 'campaign owner says we're slow' but is actually upstream load. Detection: gateway_active_streams, downstream_5xx_rate. Mitigation: drain mode + per-AZ failover; circuit-break a misbehaving tenant rather than 503'ing everyone. Cloudflare's Jun 2022 retro warns: anycast misconfig can route traffic to the wrong region — we monitor region_active_set_size and page if a tenant has >1 active region simultaneously.

Service · Campaign AcceptorGo + sqlx + librdkafka (idempotent producer)

Hosts POST /v1/campaigns (create + schedule) and PUT /v1/lists/{id}/append (chunked recipient-list staging). For create: (1) validates the template renders cleanly against a 100-recipient golden-dataset canary; (2) writes the campaign row to Postgres (UNIQUE constraint on campaign_id is the producer-side dedupe — duplicate submit returns the cached response_id); (3) registers the campaign with the scheduler tier (writes a pending_fire row in the trigger log). For list-append: streams chunks of recipients into ScyllaDB partitioned by (tenant_id, list_id, shard). Returns 201 in <200 ms for create; bulk-append is server-streaming with backpressure.

Why it exists. Considered fronting Postgres directly with a thin gateway plus a separate dedup worker. Rejected because the dedupe-then-publish ordering MUST be tight — a duplicate campaign_id submit must collapse to one row BEFORE the scheduler can pick it up, or we double-fire. Considered making the scheduler register triggers directly. Rejected because the scheduler is a sensitive Paxos cluster; exposing it to public traffic risks a CVE in the scheduler crashing the cron tier. The acceptor is the blast-radius firewall between public ingress and the scheduler.

When it fails. Postgres campaign-store down → fail-CLOSED 503 on campaign-create (better to refuse than half-create). ScyllaDB recipient-store down → list-append fails partway; the saga marks list_state='failed' via the janitor (NOT inline — saves the next caller's chunk upload from blocking on Postgres). Client retry with the same list_id lands chunks via UPSERT, so re-uploaded rows replace stale ones idempotently. The 'half-uploaded list with abandoned campaign' scenario (Critic Round 1 issue 5a): janitor reaps orphan upload state every 30 min; campaign_create rejects 'failed' lists at the API layer; if ops re-classifies a list mid-life the orphan-check worker catches the dangling campaign. Detection: campaign_create_p99 (alert >500 ms P2), list_append_chunk_throughput (alert <50 MB/s/replica P2), list_orphan_uploading_age_minutes_p99 (alert >60 min P2 — the janitor is itself the safety net here, monitor that it's running). Canary-render-failed deploy (Twitter '{firstName}' literal) prevented by golden-dataset render-diff in CI + a 100-recipient pre-flight render.

SQL · Campaign Store + Trigger LogPostgres 16 + Citus 12 (sharded by tenant_id, 16 shards)

Two logical tables: (1) campaigns(tenant_id, campaign_id, template_id, list_id, scheduled_at, priority, status) — the campaign definition; (2) trigger_log(tenant_id, campaign_id, scheduled_at, fired_at, fired_by_leader, fire_status, paxos_term) — the Borgcron-style durable log of every fire decision. Sharded by tenant_id (Citus distributed table) so a single tenant's hot campaign list doesn't bottleneck another tenant. The trigger log is the load-bearing artifact for catch-up: the scheduler reads it on failover to decide what to re-fire.

Why it exists. Considered putting the trigger log in the scheduler's local Raft state. Rejected because (a) the log must survive scheduler-tier full-cluster loss (DR scenario), (b) the catch-up controller needs to query it (SQL JOIN with delivered counts), (c) on-call needs ad-hoc 'what did we fire last Tuesday at 9 AM' queries which Raft state can't answer. Considered DynamoDB. Rejected because Citus gives us per-tenant SQL semantics + transactional pending_fire → fired state transitions with a single UPDATE — much harder in DynamoDB without conditional writes plus a status-index GSI. Postgres + Citus is what Notion (eng-blog 2023), Heap, and Cloudflare DO Resilience use for this shape.

When it fails. Per-shard primary failure → 30 s write gap for that 1/16 of tenants; reads continue from sync follower. Detection: pg_repl_lag_seconds (alert >100 ms P2), patroni_master_role per shard. Cross-shard query for catch-up (SELECT * FROM trigger_log WHERE fire_status='pending') is fan-out across all 16 shards — bounded latency ~200 ms, acceptable for the reconciler. Schema migrations use pg_repack-style online rewrites + Citus's two-phase metadata commit to avoid downtime.

Scheduler · Borgcron-styleCustom Go service + Raft (3 replicas across 3 AZ)

Leader-elected (Raft, 3 replicas) scheduler that wakes every second, reads pending_fire rows whose scheduled_at <= now() from the campaign-store, and for each one: (1) takes the lease from the lease coordinator (etcd) to assert it's still the active leader; (2) writes a firing row to the trigger log (durable write, gates everything that follows); (3) publishes a campaign.fired event to Kafka with idempotent producer enabled; (4) on producer ack, updates the trigger row to fired. On leader failover the new leader scans firing-status rows older than 30 s and DELIBERATELY re-fires them — Borgcron-style at-least-once, leaning on the per-recipient idempotency tail to make duplicates safe. Followers do NOT fire; they just shadow Raft state for failover.

Why it exists. Considered Kubernetes CronJob. Rejected per Vallery Lancey's 2020 retro — K8s CronJob stops scheduling after 100 missed runs and silently skips on concurrency conflicts. Considered Quartz clustered. Rejected because Quartz's row-lock-on-QRTZ_LOCKS approach degrades under load and its 60 s misfire threshold is too coarse for a 1-minute trigger cadence. Considered Temporal as the scheduler. Kept Temporal for the catch-up reconciler (where its workflow semantics shine) but used a dedicated Raft scheduler for the firing tier because the firing decision needs sub-second latency and direct integration with the trigger log — Temporal's queue-then-poll model adds 100–500 ms per fire. Google's Borgcron paper (Sundheim, ACM Queue 2015) is the canonical reference for this exact pattern: a Paxos-replicated leader that durably logs before firing.

When it fails. Leader dies mid-fire → up-to-5 s leadership gap; new leader replays trigger log and may re-fire rows in firing status > 30 s old (deliberate at-least-once). Trigger-log shard (Citus) primary failover (~30 s) is longer than the 1 s scheduler tick — mitigation: scheduler treats a single tick's trigger_log write failure as 'skip this tick, retry the WHOLE fire next second' (NOT block on the shard). If the shard outage exceeds the lease TTL (5 s), the scheduler voluntarily releases the lease and another replica takes over; the surviving replica's tick will also hit the same shard outage, but the catch-up controller's 5-min reconcile will absorb the missed fires once the shard recovers. Detection: scheduler_leader_count (alert >1 P1 — split-brain), trigger_fire_lag_seconds_p99 (alert >5 s P2), pending_fires_backlog (alert >100 P2, but ONLY when no shard primary is mid-failover — joint condition). Mitigation: etcd lease + fencing token; the in-flight firing row stamps fired_by_leader so the next leader can detect 'someone else was here' and not blindly re-fire. Catch-up controller is the safety net for whole-region scheduler loss. The single failure mode this whole problem exists to prevent (double-fire after partition heal) is mitigated three-deep: (a) etcd lease fences the writer, (b) trigger_log UNIQUE on (campaign_id, scheduled_at) rejects duplicate fires at the storage layer, (c) per-recipient idem-store rejects duplicate sends if (a) and (b) both fail.

Coordinator · etcd Leaseetcd 3.5 (5-node Raft, 3 AZ)

Two leases co-located on the same etcd cluster: (1) /lease/scheduler/leader — the scheduler's leader-election lease, 5 s TTL, 2 s heartbeat; the scheduler holding it is authorized to fire triggers, others stand by. (2) /lease/region/{tenant_id} — per-tenant active-region fence, 30 s TTL, 10 s heartbeat; dispatchers and catch-up workers in the active region renew it, dispatchers in the standby region check it BEFORE sending and fail-CLOSED if they don't hold it. Operator-flippable for planned failover via a typed runbook.

Why it exists. Considered DNS-based region failover. Rejected because DNS TTL + client-side resolver caching means both regions can be active for several minutes during a cutover — a guaranteed double-send window. Considered putting the lease in Postgres with SELECT FOR UPDATE. Rejected because Postgres failover (30 s) is longer than the lease TTL (5 s) and would cascade scheduler unavailability into a wider outage. Considered using Postgres advisory locks. Same problem. etcd / Consul / ZooKeeper are the right tier for sub-second leader election with linearizable reads — that's why the entire industry uses them for this exact purpose (Kafka KRaft, Kubernetes, HashiCorp Vault, every modern coordination tier).

When it fails. etcd quorum lost (3 of 5 nodes failed) → all writes fail-CLOSED → scheduler halts (no lease = no fire) and dispatchers halt (no fence = no send). Severe SPOF — we accept this over the alternative (a weaker fence) because losing 5 minutes of campaigns is recoverable via catch-up, but a 5-minute double-send window is not (TCPA + reputation damage). Detection: etcd_has_leader, etcd_proposals_committed_total, lease_renewal_lag_seconds (alert >2 s P1). Mitigation: 5-node cluster across 3 AZs absorbs single-AZ loss + 1 unrelated node; cross-region observers buy an operator-driven 10-min DR path.

NoSQL · Recipient StoreScyllaDB 5.4, RF=3, LOCAL_QUORUM, 24 nodes

Stores per-tenant recipient lists: (tenant_id, list_id, segment, recipient_id, email, locale, opt_in_proof_id, personalization_blob). Partition key = (tenant_id, list_id, segment) where segment = hash(recipient_id) % 1024 — so a single 50M list is naturally chunked into 1024 partitions of ~50K rows each, each fitting comfortably in a Scylla partition's recommended size. Clustering key = recipient_id. The fan-out controller paginates by segment for predictable parallelism: 1024 segments × ~50K recipients = 50M, fanned out by 64 concurrent workers reading 16 segments each.

Why it exists. Considered Postgres + Citus. Rejected because at 200M MAU × 5 lists avg = 1B rows, the cross-shard scan on WHERE list_id=? is expensive even with sharding; ScyllaDB's per-partition contiguous read is ~50× faster for this access pattern (Discord's 2023 Cassandra→Scylla migration retro is the precedent — they hit the exact same bottleneck on message history reads). Considered DynamoDB. Rejected because BatchGetItem is capped at 100 items, and a list with 50K items per segment would require 500 RoundTrips per segment to read fully — Scylla streams it as one range read in 200 ms. Considered S3 (recipient list as JSONL on object storage). Rejected because we need to UPDATE individual recipients (opt-out propagation, email change) and S3 doesn't support partial updates. The right pick is a wide-column store with range-scan + per-row update — that's Scylla/Cassandra.

When it fails. Single node loss → RF=3 LOCAL_QUORUM survives (W=R=2 still met); fan-out latency unchanged. Two-node loss in same AZ → some partitions go to ONE which we refuse on writes (Monzo 2019-07-29's published lesson — never relax CL=ONE on writes). Hot partition (a tenant's '100M-recipient' list creates 1024 segments of 100K each, each at the upper edge of Scylla's partition size) — mitigation: list-append API rejects appends that would push a segment over 200K rows, forces the caller to spread across more list_ids. Detection: scylla_partition_size_p99, scylla_compaction_pending, read_repair_rate. Cross-region replication lag (stream-mirroring) does NOT affect fan-out because each region runs its own recipient-store; lists are uploaded redundantly per region by the saga at campaign-create time.

Worker · Fan-out ControllerGo + Kafka consumer + Scylla driver

Consumes campaign.fired events from Kafka. For each fired campaign: (1) reads the campaign metadata + template_id + list_id from campaign-store; (2) for each of the 1024 list segments, BFS-reads ScyllaDB by segment (range scan ~5 MB / 50K rows / 200 ms); (3) for each recipient batch, produces N send.requested Kafka events keyed by (tenant_id, recipient_id) with idempotent producer enabled; (4) writes a checkpoint (campaign_id, segment, last_recipient_id, batch_acked_at) to campaign-store every 1000 recipients so resume-after-crash skips already-enqueued recipients. The controller is sticky-partitioned: one Kafka partition of campaign.fired belongs to one fanout replica, so a single campaign is owned by one worker (in-order checkpoint advancement) but 64 concurrent campaigns spread across the 32-replica fleet.

Why it exists. Considered doing the fan-out inline in the scheduler. Rejected because the scheduler's hot path is sub-200 ms; a 60-min fan-out for 50M recipients would block the scheduler tier completely (no other campaigns fire). Considered making the dispatcher do the recipient lookup per-event (no fan-out controller). Rejected because dispatchers would then read ScyllaDB once per recipient (50M random-key reads/campaign = catastrophic) vs the controller's 1024 segment range-scans. Considered Temporal for the fan-out itself. Plausible — Temporal handles long-running workflows well. We use it for catch-up (where workflow durability is load-bearing) but use a plain Kafka consumer for the fan-out tier because (a) fan-out is high-throughput stateless work that benefits from Kafka's partition-level parallelism, and (b) the checkpoint in campaign-store is durable enough — we don't need Temporal's workflow state. LinkedIn's ATC and Pinterest's NEP both run the same pattern: stateless fan-out workers + external checkpoint store.

When it fails. Single replica crash mid-fan-out → consumer-group rebalance hands the partition to another replica which resumes from the last checkpoint (worst case 5 s / 1 K recipients of re-enqueue, harmless under dedupe). ScyllaDB segment read slow (>1 s) → backs up the per-partition pipeline; mitigation: per-replica concurrency limit (16 concurrent segment reads) prevents one slow segment from blocking the whole replica. Detection: fanout_progress_recipients_per_sec (alert <80% of campaign target P2), fanout_checkpoint_lag (alert >10 K recipients P2). Kafka producer back-pressure → the controller paces its read rate to match Kafka's accept rate, never overruns the broker. The classic 'crash after enqueueing 23M of 50M' scenario (Failure Mode #3 from the spec) is exactly the case checkpointing solves — resume is bounded by the checkpoint window.

Stream · Kafka SpineApache Kafka 3.7 (KRaft mode, 24 brokers, 3 AZ)

Single Kafka cluster carrying every event the cron tier produces. Topics: campaign.fired (64 partitions, post-scheduler) → consumed by fanout; send.requested.bulk and send.requested.transactional (separate topics for bulkheading — a bulk blast must not back up a transactional 'password reset', 128 partitions each) → consumed by dispatchers; send.retry (32 partitions, post-circuit-breaker delayed retries with timestamp-keyed exponential backoff); send.delivered and send.failed (64 partitions, post-SES result, consumed by analytics-db); provider.feedback (32 partitions, post-webhook, consumed by suppression-store + analytics). Partitioned by (tenant_id, recipient_id) so per-recipient ordering is preserved (a 'pending' event precedes the 'delivered' for the same recipient).

Why it exists. Considered SQS / RabbitMQ. Rejected because (a) SQS FIFO groups cap at 300 msg/sec/group which can't carry our 14 K msg/s peak without painful sharding gymnastics, (b) replay-from-offset for catch-up backfills is impossible with SQS's 14-day point-of-no-return retention, (c) multi-consumer fan-out (dispatcher + analytics + audit all reading the same send.delivered) wants Kafka's pub-sub model, not point-to-point. Per LinkedIn ATC, Pinterest NEP, Shopify (which peaks at 66M msg/s on Kafka), DoorDash, Discord — Kafka is the production answer for this shape.

When it fails. Single broker loss → controller re-elects leaders for affected partitions, ~5 s gap on those partition writes. min.ISR=2 → single-broker loss never blocks; two-broker loss in same AZ blocks writes (correct — better than risking durability). Hot partition (one tenant blasts a single mega-list — all 50M recipients hash to nearby partitions) → mitigation: bulk topic uses (tenant_id, recipient_id) key so a single tenant spreads across all 128 partitions; per-tenant rate limit at the gateway prevents the noisy-neighbor scenario. Consumer-group rebalance during deploys is the LinkedIn KIP-429 lesson — cooperative rebalancing turned a 30–60 s STW into seconds; static membership (group.instance.id) makes restart-in-place rebalance-free.

Worker · Render & DispatchGo + handlebars renderer + SES SDK v2 + circuit-breaker

The send tail. Consumes send.requested.{bulk,transactional} from Kafka. For each event: (1) reads template from campaign-store (cached 5-min in-process, hit ratio >99%); (2) checks the active-region fence in etcd — if not ours, drops the event silently (active region will pick it up); (3) checks suppression store for opt-out / hard-bounce on this recipient — if suppressed, skips and publishes a send.failed{reason=suppressed} event; (4) checks idem-cache for a HIT on (tenant_id, campaign_id, recipient_id) — if HIT, this is a duplicate enqueue from a Borgcron re-fire or a fanout-resume; skip and publish dedupe-skipped metric; (5) on MISS, CLAIM the idem-store via Cassandra LWT (INSERT ... IF NOT EXISTS); (6) on CLAIM success, render the template with the recipient's personalization data; (7) call SES with MessageDeduplicationId = sha256(campaign_id, recipient_id); (8) on SES success, UPDATE idem-store status to sent + publish send.delivered. The render+SES call IS the critical path for the user's 'when does my email arrive' SLO.

Why it exists. Considered making this a separate render service + dispatch service. Rejected because the render → CLAIM → SES sequence is single-threaded per recipient and splitting it adds a Kafka hop (50 ms+) per send — at 14 K sends/s that's 14 K × 50 ms = 700 s of accumulated added latency per peak campaign. Considered putting rendering on the fanout controller. Rejected because rendering needs the per-recipient personalization data which lives in the recipient-store — the fanout controller already reads it for the recipient_id, but rendering also needs the template fetch and the suppression check, which happen here. Single service, sequential pipeline, is correct.

When it fails. SES 5xx storm → circuit-breaker opens, in-flight requests divert to send.retry topic with exp-backoff (1 s, 5 s, 30 s, 5 min, 30 min, 6 h, then DLQ). Workers self-throttle so the storm doesn't poison IP reputation. Detection: ses_5xx_rate (alert >2% P2, >10% P1), dispatcher_circuit_open (alert any P2), send_p99_seconds. Idem-store down → fail-CLOSED (better to delay than dup); the failed event sits in Kafka and the worker retries on next poll. Suppression-store down → fail-CLOSED on send (TCPA compliance is non-negotiable; a missed send is recoverable, a TCPA fine is not). Render error on a single template → CI golden-dataset render-diff prevents most; runtime errors land in send.failed{reason=render-error} for the campaign-owner to inspect.

Cache · Hot Dedupe (Redis Cluster)Redis 7 cluster mode, 16 shards, 1 follower/shard

Caches idem:{tenant_id}:{campaign_id}:{recipient_id} → claim_status for the SET of currently-active or recently-active campaigns (typically the last 24 h). Populated WRITE-THROUGH by the dispatcher when it WINS a CLAIM at idem-store. Read by the dispatcher on every send as the fast-path dedupe — a HIT means 'already CLAIMED in this region, skip', a MISS means 'fall through to Cassandra LWT'. Eviction is allkeys-LFU + 14-day TTL. NOT the source of truth (Cassandra is); the cache exists ONLY to take the load off Cassandra LWT for the 95% of dispatches that are first-time-sends.

Why it exists. Considered making Cassandra the only dedupe store. Rejected because Cassandra LWT (Paxos) costs ~30 ms per write and ~5 ms per read — at 14 K sends/s that's 14 K × 30 ms = 420 s of accumulated LWT latency for one campaign, plus heavy CPU on the Cassandra nodes. Two-tier dedupe is the standard pattern: hot cache for the 95% of straight-through dispatches + durable LWT for the 5% of conflicts and the eventual source-of-truth. Considered making this a per-worker in-process cache. Rejected because (a) cache hit ratio depends on cross-replica sharing — in-memory partitions by replica and gives ~2% hit rate at our 50-replica fleet, (b) a re-fire from Borgcron failover MUST hit a shared cache to dedupe, not a per-replica one.

When it fails. Single shard down → 1/16 of dispatches fall through to Cassandra LWT (5–10× slower, ~30 ms/op vs ~1 ms/op) — degraded but correct; Cassandra absorbs the spike at ~10× the cost. Full cluster restart → ALL dispatches fall through; Cassandra LWT saturates within seconds — page on idem_cache_hit_ratio < 0.85. Mitigation: cluster-mode auto-failover handles single-shard within 15 s; full-cluster restart is rare (last incident: never in this design's history). Hot key: a single campaign's recipients all carry (tenant_id,campaign_id) hash tag → cluster pins them to ONE slot pair → single shard absorbs 100% of the campaign's lookups. Mitigation: at >10K sends/s on a single campaign, the dispatcher salt-shards by (tenant_id, campaign_id, recipient_id % 16) instead of by (tenant_id,campaign_id) (per Failure Mode #5 in the spec). NEVER trust this cache for opt-out semantics — those go straight to suppression-store via LWT.

NoSQL · Idempotency Store (LWT)Cassandra 4.1, RF=3, LOCAL_QUORUM + LWT (Paxos) on CLAIM

Source of truth for per-recipient dedupe. Schema: PK (tenant_id, campaign_id, recipient_shard) where recipient_shard = hash(recipient_id) mod 8, CK recipient_id, columns (claim_status: claimed|sent|failed|suppressed, claimed_at, claimed_by_pod, ses_message_id, attempt_count, ttl). The dispatcher CLAIMs a recipient via Cassandra LWT: INSERT INTO idem (...) VALUES (...) IF NOT EXISTS — Paxos-coordinated linearizable write. Wins → row inserted, dispatcher proceeds with SES call. Loses (applied=false) → row already exists, dispatcher dedupes the send. On SES success, dispatcher UPDATEs claim_status='sent', ses_message_id=? (no LWT on the update, only the initial CLAIM). 14-day TTL on every row. The recipient_shard mod 8 in the PK is the load-bearing decision: without it, a single 50M-campaign first-fire lands 14 K LWT/s on ONE Cassandra partition coordinator and exceeds the ~5 K/s/partition Paxos ceiling by 2.8×; with mod 8, the load spreads across 8 partition groups (~1.7 K/s each, comfortable). This is not a 10× scale fix — it is needed on day one and the Critic Round 1 review flagged the original single-partition design as broken under first-fire.

Why it exists. Considered Redis SETNX as the source of truth. Rejected because Redis async failover has RPO=5 s — 5 seconds of dedupe state vanishing on AZ outage is enough to leak millions of dupes for an in-flight 50M campaign (Failure Mode #4 in the spec). Considered DynamoDB conditional PutItem. Plausible — AWS Builders' Library ARC403 endorses this exact pattern. We picked Cassandra over DynamoDB because (a) catch-up controller needs RANGE SCANS over (tenant_id, campaign_id, recipient_shard) to compare intended vs delivered — Cassandra's clustering key gives O(log N) range read across the 8 shards (8 small scans, parallelized); DynamoDB needs a GSI or N×GetItem; (b) the team already runs Cassandra for suppression — operational compounding. Both Cassandra-LWT and DynamoDB-conditional-PutItem are defensible; the engineering trade is operational topology vs managed-service convenience.

When it fails. Single node loss → RF=3 LWT survives (Paxos quorum is 2 of 3); CLAIM latency unchanged. Two-node loss in same AZ → some partitions have only ONE replica, CLAIM blocks (correct — better than risking dupe). 8-way recipient-shard prevents the single-partition hot-coordinator problem on first-fire (1.7 K LWT/s/partition vs ~5 K/s/partition ceiling). Orphan claimed rows from a dispatcher crash between CLAIM-win and SES-success — janitor reclaims every 5 min by re-publishing to send.retry; the second dispatcher gets applied=false on its CLAIM (the row still exists) and proceeds to SES with the existing CLAIM, no double-send. Detection: lwt_p99_seconds (alert >100 ms P2), lwt_paxos_contention_rate (alert any P2 — applied=false is NOT a failure, only LWT timeouts/errors count), cassandra_pending_compactions, idem_claimed_orphan_age_seconds_p99 (alert >10 min P2). Cross-region replication via async stream-mirror; on planned failover, catch-up does cross-region range-scan before re-dispatch so already-sent recipients are skipped (~80 ms p99 cross-region read penalty, paid once per failover not per send).

NoSQL · Suppression (TCPA-grade)Cassandra 4.1, RF=3, LWT on opt-out writes

Source of truth for 'do NOT send to this recipient'. Three categories: (1) opt-outs — user clicked unsubscribe (CAN-SPAM requires processing within 10 business days, we do it inline within seconds); (2) hard-bounces — recipient address doesn't exist (SES feedback); (3) complaints — recipient flagged as spam (SES feedback). Schema: PK (channel, identifier) where channel='email' and identifier is the lowercased email address. Opt-out writes use Cassandra LWT (INSERT ... IF NOT EXISTS) for linearizable cross-region consistency. Reads on the send path use LOCAL_QUORUM (W=R=2) — strong within region, sufficient because regional ownership ensures the recipient was last seen by the home region.

Why it exists. Considered keeping suppression in Postgres + Citus alongside campaigns. Rejected because suppression reads are on the send hot path (14 K/s peak per campaign, 1 read per send → 14 K reads/s sustained per active blast), and Citus's cross-shard fan-out for WHERE email=? is wasteful for a point-lookup. Considered making it a Redis-only cache. Rejected because TCPA statutory damages are $500–$1500 per opt-out violation — the cost of a cache miss on opt-out is too high; the source of truth must be durable with linearizable opt-out writes. Considered DynamoDB. Plausible — but we already run Cassandra for idem-store, and LWT semantics are identical.

When it fails. Suppression-store down → dispatcher fail-CLOSED on send (a missed send is recoverable, a TCPA violation is $46 K). Detection: suppression_check_p99 (alert >50 ms P2), suppression_lwt_timeout_or_error_rate (alert any P1 because that means opt-outs are not landing — direct legal exposure). NOTE: applied=false from LWT (the recipient already opted out) is NOT a failure and is excluded from this metric — alerting on raw applied=false would alert-storm during every mass-unsubscribe event. Hot key (mass-unsubscribe campaign) — mitigation: opt-out clicks are rate-limited at the unsubscribe-handler service to <1 K/s per tenant, so suppression LWT contention is bounded. Cross-region LWT failures during a region partition → opt-out writes block (correct — better to delay confirmation than to silently lose the opt-out).

External · Amazon SESAmazon SES (per-region sub-accounts + dedicated IP pools)

Third-party email delivery. We hold multiple SES configurations: (a) transactional sub-account in us-east-1 with a dedicated IP pool warmed to the 50 K/s tier — for 'password reset' / 'order shipped' / '2FA' traffic; (b) marketing-bulk sub-account in us-east-1 with a separate dedicated IP pool warmed to the 14 K/s tier — for the 50M-recipient blasts; (c) marketing-bulk-west sub-account in us-west-2 with its own dedicated IP pool — for regional failover and load-spreading on truly massive campaigns. Each call: SES validates SPF/DKIM/DMARC alignment, scans for spam-triggering content via its content classifier, accepts the message, returns a MessageId, then handles MTA-level retries and bounce processing on its side. Bounces and complaints land at our webhook-rx via SES → SNS → HTTPS.

Why it exists. Considered self-hosted Postfix + IP block. Rejected because IP-reputation management is a full-time team — SES + SendGrid + Mailgun do it for us, and recovery from a reputation hit takes weeks. Considered SendGrid as primary. Plausible — SendGrid's published 1.5M/hr ≈ 416/s sustained is workable; we keep SendGrid as the secondary provider for failover. SES wins as primary because it's the AWS-native cheapest option at scale ($0.0001/email).

When it fails. SES 5xx storm (regional API degradation) → dispatcher circuit-breaker diverts to send.retry topic with exp-backoff; campaign throughput drops; if storm lasts >30 min, the catch-up controller will detect (delivered count vs intended count diverges) and pace replay. Bounce rate spike (>5%) → SES auto-pauses the sub-account; we get a Reputation.BounceRate CloudWatch alarm. Mitigation: pre-flight list hygiene (suppress hard-bounces from previous campaigns), automated pause when bounce rate >3%, sub-account isolation prevents whole-tenant outage. Detection: ses_throttle_rate (alert >2% P2), ses_reputation_bounce (alert >3% P1), ses_reputation_complaint (alert >0.05% P1). 429 backoff cascade (Failure Mode #6) — AIMD token bucket per-pool plus circuit-breaker open is the documented mitigation.

Service · Webhook Receiver (SES)Go + AWS SDK v2 (SNS-signed payload verify)

Receives bounce, complaint, and delivery webhooks from SES via SNS HTTPS POST. For each: (1) verifies the SNS signature (X.509 cert validated against AWS's published cert URL, cached); (2) parses the payload (bounce_type, recipient, original_message_id); (3) dedupes against (provider, provider_event_id) in idem-cache; (4) publishes a provider.feedback event to Kafka; returns 200 OK in <50 ms (SES retries on non-200 with backoff up to 24 h). Does NOT do the suppression-store write inline — that happens async in the suppression-update worker that consumes provider.feedback. Keeping this tier minimal absorbs bounce storms without backing up.

Why it exists. Considered having the dispatcher poll SES for delivery results. Rejected because SES doesn't support polling at scale — webhook is the only practical feedback channel. Considered making the webhook receiver write suppression-store synchronously. Rejected because a bounce storm (12M bounces in 10 minutes after a deliverability hit — Failure Mode #13) would back up Cassandra LWT and 503 the webhook, which would cause SES to retry, which would compound the storm. Decoupling via Kafka means the webhook is a thin durable hand-off; suppression updates drain at their own pace.

When it fails. SES webhook payload validation fails (cert rotation issue, malformed JSON) → log + return 400 (do NOT 500, or SES retries forever). Kafka publish fails → 500 to SES → SES retries with backoff. Webhook storm (bounce campaign hits 12M bounces in 10 min) → consumer-side backpressure absorbs via Kafka buffering — webhook-rx itself stays p99 <50 ms because all it does is verify + publish. Detection: webhook_p99 (alert >100 ms P2), webhook_5xx_rate (alert >1% P1), webhook_dedupe_hit_rate. Mitigation: SES has its own 24-h retry budget; a brief webhook-rx outage is recoverable via SES's retry.

Worker · Catch-up Reconciler (Temporal)Temporal 1.22 workflows + custom diff worker

Long-running Temporal workflow per campaign. Triggered (a) on a schedule every 5 min for in-flight campaigns; (b) on-demand by ops after a known outage (scheduler down, region failover, SES outage). Algorithm: (1) acquire the catch-up lease in etcd at /lease/catchup/{campaign_id} (prevents two workflows reconciling the same campaign); (2) read intended recipient count from campaign-store (SELECT segment, recipient_count FROM list_segments WHERE list_id=?); (3) read delivered/claimed state DIRECTLY from idem-store via range-scan across all 8 recipient_shard partitions for the campaign (SELECT recipient_id, claim_status FROM idem WHERE (tenant_id, campaign_id, recipient_shard) IN (?, 0..7)) — idem-store is the source of truth for 'did we send this'; analytics-db is informational only and is NOT read on the reconciler path (Critic Round 1 issue 6c: analytics-db lag would false-positive-trigger a send.retry storm — direct idem-store read avoids the circular dependency); (4) for each recipient with claim_status NOT IN ('sent', 'suppressed', 'failed'), publish a send.retry event with attempt_origin='catch-up' — at a paced rate of ≤ 50% of baseline send rate (so SES isn't overwhelmed); (5) when intended ≈ (sent + suppressed + failed), mark campaign complete. Workflow can resume from any step on worker crash (Temporal's durable workflow state).

Why it exists. Considered making the scheduler do catch-up. Rejected because scheduler hot path is sub-200 ms; catch-up scans take minutes to hours and don't belong in that latency budget. Considered a cron-triggered batch job. Rejected because catch-up needs to resume mid-scan after a worker crash (a 50M-recipient diff takes hours to complete), and a plain cron job has no durable state — it would restart from scratch on every crash. Temporal's workflow durability is the right abstraction: write step-by-step state to Temporal's history, resume on crash, retry per-step. DoorDash's published 'Cadence as fallback' post is the precedent for using a workflow engine for exactly this 'safety-net reconciler' role.

When it fails. Catch-up workflow stuck waiting on a slow analytics-db query → Temporal's per-activity timeout fires, retries on a different worker, eventually escalates to ops via a P2 alert. Two catch-up workflows for the same campaign (lease race) → etcd lease prevents (whichever workflow gets the lease wins; the other waits or fails-fast). Catch-up storm (Failure Mode #9 in the spec) — pacing prevents; if pacing is mis-configured, the SES throttle + circuit-breaker absorbs the secondary storm. Detection: catchup_workflow_age_seconds (alert >2 h P2), catchup_lag_recipients (alert >100 K behind expected P2), catchup_relevance_drops (informational metric for the campaign-owner UI). Temporal cluster down → catch-up halts but happy-path is unaffected; ops can replay missed catch-ups manually from trigger_log.

Analytics DB · ClickHouseClickHouse 24.x (replicated, 12 nodes, ZooKeeper-Keeper)

Consumes send.delivered, send.failed, and provider.feedback from Kafka via the ClickHouse Kafka engine. Stores per-event rows: (event_time, tenant_id, campaign_id, recipient_id, segment, channel, provider, message_id, status, error_code, attempt_count). Materialized views aggregate to: (a) per-campaign delivered/failed/bounced/complained counters (powering the campaign-owner UI); (b) per-tenant rolling bounce rate (driving the SES reputation alarm); (c) per-segment delivery progress (informational input to the catch-up controller's freshness check — the reconcile diff itself reads idem-store). 30-day hot retention, 4-year cold retention via S3 + Parquet for CAN-SPAM audit.

Why it exists. Considered Postgres for analytics. Rejected because aggregate queries over 50M-recipient × 4K-campaign tables are O(N) scans that crush an OLTP engine; ClickHouse's columnar engine answers them in ~100 ms vs minutes. Considered putting analytics in Snowflake / BigQuery. Plausible for the cold-archive tier; we use S3 + Parquet because we already export there for compliance, and ClickHouse can query S3 Parquet via its s3() function for the rare cold-archive query.

When it fails. ClickHouse node loss → reads fail over to replica within 60 s; ingest pauses on the affected shard for 30 s. Kafka engine resumes from the last committed offset on restart, so no data loss for ingest. Cardinality storm if a campaign accidentally includes recipient_id as a dimension on a metric (Failure Mode #20) — mitigation: per-tenant series limits at the ClickHouse ingest layer, plus we explicitly forbid recipient_id as a dimension at the metric library level. Catch-up controller depends on analytics-db reads being fresh; a 60-s lag is acceptable, anything beyond → catch-up workflow retries until fresh.

Tracing · OTel CollectorOpenTelemetry Collector + Tempo backend

Receives OTLP spans from scheduler, fanout, dispatcher, webhook-rx, catch-up workflows. Per-trace context propagated via traceparent header on every Kafka message (in the message headers, not the body — keeps deserialization independent of tracing). Stored 7 days hot in Tempo, sampled to 4 weeks for high-cardinality investigations. Powers Slack's tracing-notifications pattern (eng-blog 2017): a customer support agent pastes recipient_id + campaign_id into a forensics UI and gets back the full timeline — fire → fanout → dispatch → SES → delivery (or where it went wrong).

Why it exists. Considered relying on logs alone. Rejected per Slack's 2017 post — for fan-out systems, logs without trace correlation are unsearchable at scale; on-call spends 30+ min reconstructing 'what happened to recipient X' from 6 different services' logs. Tracing makes this a 30-second lookup. Considered head-based sampling at 1% to control cost. Rejected because we need 100% sampling on send.failed and the catch-up workflows — tail-based sampling at the OTel collector keeps the per-trace decision local while still keeping the rare-and-interesting traces.

When it fails. Tracing backend unreachable (Tempo down, network issue) → OTel collector buffers 2 min locally, then drops with a tracing_spans_dropped_total metric (NOT a page — observability is degradable). NEVER on the sync send path (OTel emit is async UDP). Detection: tracing_export_failure_rate (informational P3), tracing_collector_buffer_pct. Critically — we do NOT page on tracing-down during an incident because the same root cause may be taking down monitoring (the Failure Mode #20 scenario: cardinality death takes down both tracing and Prometheus simultaneously). Mitigation: self-monitor tracing via an INDEPENDENT lightweight metrics pipeline.