Distributed Counter

Problem statement (interviewer prompt)

Design a high-throughput distributed counter for things like like / view counts: 1M+ increments/second across millions of distinct counters, sub-second reads, eventually consistent OK. Avoid hot-key bottlenecks via sharded counters or probabilistic estimation (HLL).

flowchart LR
  E[Event source]
  SHARD[Sharded counter<br/>per key shard]
  AGG([Aggregator])
  API[Read API]
  E --> SHARD --> AGG --> API

    classDef client fill:#dbeafe,stroke:#1e40af,stroke-width:1px,color:#0f172a;
    classDef edge fill:#cffafe,stroke:#0e7490,stroke-width:1px,color:#0f172a;
    classDef service fill:#fef3c7,stroke:#92400e,stroke-width:1px,color:#0f172a;
    classDef datastore fill:#fee2e2,stroke:#991b1b,stroke-width:1px,color:#0f172a;
    classDef cache fill:#fed7aa,stroke:#9a3412,stroke-width:1px,color:#0f172a;
    classDef queue fill:#ede9fe,stroke:#5b21b6,stroke-width:1px,color:#0f172a;
    classDef compute fill:#d1fae5,stroke:#065f46,stroke-width:1px,color:#0f172a;
    classDef storage fill:#e5e7eb,stroke:#374151,stroke-width:1px,color:#0f172a;
    classDef external fill:#fce7f3,stroke:#9d174d,stroke-width:1px,color:#0f172a;
    classDef obs fill:#f3e8ff,stroke:#6b21a8,stroke-width:1px,color:#0f172a;
    class E,SHARD,API service;
    class AGG compute;

flowchart TB
  subgraph Write[Increment path]
    EVT[Event sources]
    SHARD[N micro-counters per key<br/>shard by event hash]
    LOCAL[Local batch + flush]
    REDIS[(Redis counters per shard)]
    EXACT[(Durable per-shard ledger)]
  end

  subgraph Read[Read path]
    AGG([Aggregator: sum across shards])
    CACHE[(Approx cache 1s-10s)]
    API[Counter API]
  end

  subgraph Approx[Approximate fallback]
    HLL[HyperLogLog uniques]
    CMS[CMS hot keys]
  end

  subgraph Recon[Reconciliation]
    BATCH[Batch totals - exact]
    DRIFT[Drift alarm]
  end

  Write --> Read
  Approx --- Write
  Recon --- Write

    classDef client fill:#dbeafe,stroke:#1e40af,stroke-width:1px,color:#0f172a;
    classDef edge fill:#cffafe,stroke:#0e7490,stroke-width:1px,color:#0f172a;
    classDef service fill:#fef3c7,stroke:#92400e,stroke-width:1px,color:#0f172a;
    classDef datastore fill:#fee2e2,stroke:#991b1b,stroke-width:1px,color:#0f172a;
    classDef cache fill:#fed7aa,stroke:#9a3412,stroke-width:1px,color:#0f172a;
    classDef queue fill:#ede9fe,stroke:#5b21b6,stroke-width:1px,color:#0f172a;
    classDef compute fill:#d1fae5,stroke:#065f46,stroke-width:1px,color:#0f172a;
    classDef storage fill:#e5e7eb,stroke:#374151,stroke-width:1px,color:#0f172a;
    classDef external fill:#fce7f3,stroke:#9d174d,stroke-width:1px,color:#0f172a;
    classDef obs fill:#f3e8ff,stroke:#6b21a8,stroke-width:1px,color:#0f172a;
    class EVT,SHARD,LOCAL,API,HLL,CMS,BATCH,DRIFT service;
    class EXACT,CACHE datastore;
    class REDIS cache;
    class AGG compute;

Why shard a counter

• A single hot key (e.g. tweet likes for a celeb) bottlenecks updates.
• Split into N counters; sum on read.
• Trade-off: read fan-out vs write contention.

Approximate vs exact

• For "views/likes at scale" approximate (HLL, CMS) is fine.
• Money counters must be exact → use ledger + read-side cached aggregate.

Glossary & fundamentals

Concepts referenced in this design. Each row links to its canonical page; the tag column shows whether it is a high-level (HLD) or low-level (LLD) concept.

Tag	Concept	What it is	Page
HLD	Sharding	horizontal partitioning across nodes	database-sharding
HLD	Probabilistic data structures	Bloom, HLL, Count-Min, MinHash, t-digest	probabilistic-data-structures
HLD	Event sourcing + CQRS	commands -> events; separate read model	event-sourcing-cqrs

Quick reference

Functional

• Increment by key (likes, views, votes).
• Read current count.
• Optional decrement.

Non-functional

• p99 increment < 10 ms.
• Reads cached for short windows.
• Approximate is acceptable for non-financial use.

Trade-offs

• Sharded micro-counters scale writes; sum-on-read is the cost.
• Approximate with HLL/CMS for billion-scale.
• Exact requires ledger + reconciliation.

Refs

• Facebook "TAO" counters.
• Discord counters posts.
• ByteByteGo "Design distributed counter".

FAQ

How do you design a distributed counter?

Shard the counter into N micro-counters per key and have writers increment a random shard. A background aggregator periodically sums shards to a canonical total readers can fetch.

Why split a counter into shards?

A single hot key serializes writes on one node. Sharding spreads increments across many nodes so total throughput scales horizontally without per-key contention.

What is a CRDT counter?

A G-counter or PN-counter is a CRDT where each replica owns a slot. Merge is element-wise max or sum, so counters converge across replicas without coordination.

When use HyperLogLog instead of an exact counter?

Use HLL when you need cardinality (unique users) at scale within a couple percent error and a few KB of memory. Exact counters need much more storage at billions of items.

How accurate are sharded counters on read?

Reads on the aggregate may lag actual writes by the flush interval. For UI counters this is fine; for billing use a strongly consistent counter or a transactional log instead.