Design a distributed counter

I would ask before drawing anything. Does the count have to be exact, or is "within a second or two, never going backwards" enough? Is anyone billed from it? Can counters go down (an unlike)? A like counter that also enforces one like per user needs a set of (user, item) pairs, which is a different problem from a view counter that only adds. How many counters are there, how skewed is the traffic, and are reads global or per region?

Assumed scope: 100 million counters, add-only plus an optional decrement, eventual consistency with a delay of a few seconds, three regions. In scope: ingest, storage, hot keys, reads, the multi-region merge. Out of scope: per-user uniqueness (a follow-up), bot filtering, slicing counts by dimension.

These are planning figures for the exercise, not facts about any company.

• Write rate: 1 billion increments a day ÷ 86,400 s is about 11,600 a second; at a 5× peak, 58,000 a second.
• Against one primary: a relational primary is a planning figure of 5,000 committed writes a second. 58,000 ÷ 5,000 = 11.6, so one row per increment on one box is out. Sixteen shards puts each at 58,000 ÷ 16 = 3,625 a second.
• Hot key: one item taking 20,000 a second is a single row that every writer must update in turn. Sharding by key does not help, because the load is one key. Batching does: collapse a second of increments into one delta and that row sees 1 write a second instead of 20,000.
• Reads: assume 5 reads per write at peak, 58,000 × 5 = 290,000 a second. That is 14.5 times the 20,000 simple-read figure for one box, so reads need a cache. At a 99% hit rate the store sees 290,000 × 0.01 = 2,900 a second.
• Size: 100 million counters × 16 B (id plus count) = 1.6 GB; with roughly 3× overhead, about 5 GB. Capacity is not what forces sharding. Write rate and hot keys are.
• Cross-region: US to Western Europe is 70 to 90 ms. A synchronous global increment would add that to every write, so each region counts locally and merges later.

The numbers force: a durable log in front, in-memory batching, a sharded store, a cache for reads, and region-local counting.

The shape is: API appends to a log, one aggregator per partition sums and writes to the matching store shard, reads go through a cache.

1. Ingest. The API validates and appends {counter_id, delta, event_id} to a partitioned log, then acknowledges. The partition is hash(counter_id) mod 16. The log is the source of truth: a lost store is rebuilt by replay, and a traffic spike lands in the log, not on the store. The hot key at 20,000 events a second is about 1 MB/s at 50 B each, far under the roughly 10 MB/s per-partition planning figure.
2. Aggregate. One consumer owns each partition. It keeps a map of counter id to pending delta and flushes every second. A flush is one transaction on the one store shard that matches its partition: UPDATE counters SET n = n + :d, version = version + 1 WHERE id = :id for each counter, plus UPDATE offsets SET off = :o WHERE partition = :p. Because the log partition and the store shard use the same hash, the offset row lives in the same shard and commits atomically with the deltas.
3. Replay safety. On restart the consumer reads its stored offset and resumes after it, so a replayed batch is never applied twice. The idempotency key is the (partition, offset), not the individual event. A client retry of the API call is covered by event_id, de-duplicated in the aggregator for a short window.
4. Store. A table counters(id, n, version), 16 shards, primary key id. Increments are always n = n + delta in the database. Never read the value, add in the application and write it back, because two writers will lose an update.
5. Read. API, then cache (TTL 2 s with plus or minus 20% jitter), then the shard. A cache fill carries the row version and never replaces a newer version with an older one, so a reader does not see the number go backwards. For a hot key the aggregator can push the new value into the cache on each flush instead of waiting for expiry.

Hard problem 1: the hot key. Batching first, as above: 20,000 increments become one delta a second per aggregator. If a single key ever outgrows one partition and one row (hundreds of thousands a second), detect it at runtime and split it into 16 slots. Producers route that key's events by (counter id, random slot) so they spread over partitions, which batching alone cannot do, and each slot row counter_slots(id, slot, n) lives on the shard that matches its partition. A read sums 16 rows and the result is cached. Only hot keys get slots, because every read of a slotted key costs 16 reads.

Hard problem 2: regions. Each region runs the whole pipeline and owns one row per counter: (counter_id, region, n). A region only ever writes its own row, so there are no write conflicts. Rows replicate asynchronously to the other regions, and the global value is the sum of the latest known row from each region. Each row only grows, so a merge is "take the larger", which is a grow-only counter. A decrement gets a second column and the value is increments minus decrements. Staleness is the replication delay, usually about a second, and it can never invent a count.

Failure. If an aggregator dies, another takes its partition and resumes from the stored offset. If the cache dies, the store absorbs 290,000 reads a second for a moment, so protect it with a per-key lock on refill (one loader per key) and a short serve-stale window.

• Durable log versus batching in API memory. Pick the log when the count is billed or ranks something, because a crashed API node loses its unflushed second. Flip to in-memory batching for plain view counts where losing a second of increments is acceptable and the extra hop is not worth it.
• Direct database increment versus log plus aggregator. Direct is simple and exact at read time, fine up to a couple of thousand writes a second. Pick the log past the 5,000-a-second planning figure or when a hot key appears. Flip back when volume is small and a read must see the very latest write.
• Long cache TTL versus short. Pick 2 seconds: it removes 99% of reads and the number is only seconds old. Flip to push-on-flush for the few hot keys where even 2 seconds looks wrong, or to a longer TTL for cold counters.
• Slots versus a single row. Pick the single row with batching first. Flip to slots when one key's raw event rate exceeds what a single partition and its aggregator can take, accepting 16 times the read cost for that key only.

The common mistakes on this question:

• Reading the value, adding one in application code and writing it back. It loses updates under concurrency; the add belongs in the store.
• One row per counter taking every increment directly, so a single viral item serialises 20,000 writers.
• Using an in-memory INCR with no story for a restart. If the instance dies, the counts since the last snapshot are gone.
• Retrying an increment after a timeout with no event id, which double counts. The outcome of a timed-out call is unknown.
• Synchronous cross-region agreement for every increment, paying 70 to 90 ms on the write path.
• Claiming "exactly once" without saying that it is at-least-once delivery plus an idempotent apply.