Sharded Counters

Sharded counters

Count a thing that is incremented far faster than a single row, key, or partition can serialize writes — likes, views, votes, rate tallies, inventory decrements. The trap is the hot counter: every writer contends on one record, so latency climbs and throughput plateaus no matter how big the box is. Getting it wrong turns a trivial +1 into the bottleneck of the whole feature.

When to reach for this

Concurrent increments to a single logical count exceed what one row/key can absorb — a viral post's like count, a live-event view counter, a global rate tally. The symptom is write contention (lock waits, CAS retries, partition hot-spotting) on one record while the rest of the store is idle. Reaching for this means the write side is the problem, and an exact-to-the-millisecond total is not required.

When NOT to

Low write rate (a single atomic INCR handles thousands/sec — don't shard a counter nobody is hammering; YAGNI). Counts that must be transactionally exact and read-after-write consistent at every instant (bank balances, seat inventory at sell-out) — that's a transactional decrement, see consistency-coordination, not a fan-out tally. Counting distinct items exactly (unique visitors) where you also need the member list — that's a set in the store, not a counter. If reads dominate and writes are cheap, you need a cached aggregate, not sharding.

Clarify first

• Write rate to the hottest single count — peak increments/sec on one logical counter, not the aggregate (→ back-of-the-envelope).
• Exact or approximate — is an off-by-a-few total acceptable, and for how long may shards disagree (eventual)? Drives shard count and read path.
• Counting occurrences or distinct items — a running total vs. unique-count (likes vs. unique viewers) decides plain shards vs. HyperLogLog.
• Read rate and freshness — how often is the total read, and how stale may the served number be (sub-second? minutes?).
• Time-windowed or lifetime — "views in the last hour" needs bucketed keys and expiry; a lifetime total does not.

The options

• Single atomic counter — one row/key with atomic INCR/UPDATE +1. Use when peak write rate on the hottest count is well within one node's serialized write throughput. The default; don't outgrow it prematurely.
• Write-sharded (striped) counter — split one logical count into N physical shards (counter:{id}:shard:{0..N-1}); each write increments a random/hashed shard, reads sum all N. Use when single-key contention is the bottleneck and the total may be eventually consistent.
• Approximate distinct count (HyperLogLog) — a fixed-size probabilistic sketch (~12 KB) that counts unique items with ~2% error. Use for uniques at scale where exact membership isn't needed (unique visitors, distinct search terms).
• Time-windowed (bucketed) counters — key the counter by time bucket (views:{id}:2026-06-01T14), increment the current bucket, sum recent buckets on read, expire old ones. Use for "last N minutes/hours" rate-style counts.
• Aggregate-on-read + cached total — sum shards (or roll up) periodically and serve the cached number. Use when reads vastly outnumber writes and a slightly stale total is fine (pairs with caching).

Trade-offs

Behavior under stress

A counter is a tiny thing that punches above its weight in an outage.

• Hot-shard skew: if writes pick shards by hash(userId) instead of random, one viral actor or a bad hash can still pile onto one shard. Mitigate: pick the shard at random per write; size N to peak contention, not average.
• Read amplification on spikes: when a count goes viral, reads of the total multiply the N-shard sum across the read fan-out and can overload the store. Mitigate: cache the aggregate and refresh on an interval, not per read (→ caching).
• Lost increments: fire-and-forget increments (or a crash before flush in a buffered/write-back path) silently undercount. Mitigate: use the store's atomic increment, accept the eventual-consistency window explicitly, and reconcile from a source of truth if exactness later matters.
• Window-boundary stampede: time-bucketed counters all roll to a new key at the top of the hour — a synchronized cold bucket plus a flood of reads. Mitigate: pre-create buckets and jitter rollups.
• Mass expiry: bucketed counters expiring together can spike the store. Mitigate: stagger TTLs.

Monitor: per-shard write distribution (skew), increments/sec vs. node ceiling, read-path latency for the N-sum, sketch error budget (HLL), and under/over-count drift against any source of truth.

How to apply

1. Clarify the inputs — peak increments/sec on the hottest single count, exact-vs-approximate tolerance, occurrence-vs-distinct, read rate, and freshness budget (see Clarify first). If no number shows one count is too hot, stay on a single atomic counter (YAGNI).
2. Pick from the trade-off table — single atomic if it fits one node; write-sharded if single-key contention is the wall; HyperLogLog for uniques; time-buckets for rolling windows. Combine (e.g. sharded + cached aggregate).
3. Set the key knobs — choose N (shard count) from peak contention, the shard-selection rule (random, not user-hashed), the read aggregation method, bucket granularity + TTL for windows, and the cache refresh interval.
4. Stress-test the choice — walk Behavior under stress: confirm shard skew, read amplification, lost increments, and window boundaries each have a mitigation the traffic profile actually needs.
5. Size it with numbers — N ≥ peak increments/sec ÷ per-shard write ceiling; confirm the N-sum read cost and any HLL error fit the budget (→ back-of-the-envelope).
6. Pick a provider — default to the generic recipe; open a provider file only if the user named a cloud (see Choosing a provider).
7. Show the trade-off in the answer — give a Decision | Solves | Worsens | Change it when… table with one row per choice you made here (e.g. the option, its store, its failure policy); the last column is a concrete trigger (a number or observable signal). Then name what breaks first.

Dos and don'ts

Do

• Start with a single atomic counter; shard only when a number shows one count is the bottleneck.
• Pick the write shard at random so load spreads evenly regardless of the actor.
• Cache the summed aggregate and refresh on an interval when reads dominate.
• Use HyperLogLog for uniques-at-scale and state the ~2% error as a known cost.
• Expire time-bucket keys with staggered TTLs and pre-create the next bucket.
• State the eventual-consistency window out loud — sharding trades exact-now for throughput.

Don't

• Don't shard a counter that a single INCR already handles (premature sharding adds read cost for nothing).
• Don't hash the shard by user/entity ID — a hot actor reconcentrates the load.
• Don't sum N shards on every read of a viral count — cache the aggregate.
• Don't use a sharded/eventual counter where the number must be transactionally exact (money, sell-out inventory) → consistency-coordination.
• Don't reach for HyperLogLog when you also need the member list or an exact count.

Numbers that matter

The deciding figure is peak increments/sec on the single hottest count vs. a node's serialized write ceiling — that ratio sets N. A HyperLogLog sketch is ~12 KB for billions of uniques at ~2% standard error, regardless of cardinality — the reason it beats a stored set at scale. Reading a sharded total costs N lookups, so N trades write headroom for read cost. For QPS rates, per-node write ceilings, and storage sizing, don't restate them here — see back-of-the-envelope.

Interface sketch

A sharded counter is a key contract, not one value:

• Write: INCR counter:{id}:shard:{rand(0..N-1)} (atomic, fire to one shard).
• Read: sum(GET counter:{id}:shard:{0..N-1}) — or read the cached aggregate count:{id} refreshed every T seconds.
• Distinct: PFADD uniq:{id} {member} then PFCOUNT uniq:{id} (HLL sketch).
• Windowed: INCR views:{id}:{bucket} with a TTL; read sums the recent buckets.

Decide N, the shard-selection rule, the aggregation/refresh policy, and the window granularity up front — they are the contract, not implementation details.

Choosing a provider

Default to the generic recipe above. If the user names a cloud, read references/providers/<provider>.md for the managed-service mapping, quotas/limits, and provider-specific trade-offs. If no file exists for that provider, the generic recipe is the answer.

Diagram

To visualize the fan-out write path (writer → random shard) and the aggregate-on-read sum (read → N shards → cached total), use the in-plugin architecture-diagram skill — shards share the store color, the read fan-out is a dashed sum arrow, and the cached aggregate sits in the cache color.

Related building blocks

• data-storage — depends on this for where the shards physically live; sharding/partitioning theory and key design are owned there.
• caching — pairs with this to serve the cached aggregate so a viral count's reads don't re-sum N shards every time.
• consistency-coordination — depends on this for the exact-vs-eventual count decision; transactional/atomic semantics and quorum are owned there.
• back-of-the-envelope — feeds into this: it supplies the per-count write rate and per-node ceiling that justify sharding and set N.
• system-design — owned-concept lives in the orchestrator: the reasoning loop, the trade-off method, and the ten failure modes.

References

• references/deep-dive.md — shard-count math, random vs. hashed selection, HyperLogLog mechanics and error, time-bucket layout, roll-up/aggregation patterns, and reconciliation. Read when designing the counter in detail.
• references/providers/{generic,aws,azure,gcp}.md — service mappings, atomicity/contention limits, and pitfalls per environment.