/sharded-counters

name: sharded-counters
description: This skill should be used when the user needs a "sharded counter", "distributed counter", to "count likes / views at scale", handles a "high-write counter" or "hot counter contention", asks about "approximate counting", "real-time counts", or "HyperLogLog". It gives the recipe for absorbing write-heavy counting without a single hot row. Use it whenever one row/key takes concurrent increments faster than it can serialize them, even if the user doesn't say "sharded counter".

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

| Option | What it solves | What it worsens | Change it when | |---|---|---|---| | Single atomic counter | Simplest; exact; read-after-write trivial | One hot record caps write throughput; contention under spikes | Increments on one count exceed one node → shard the writes | | Write-sharded counter | Spreads write load N-way; removes the hot spot | Reads cost N lookups + sum; total is eventually consistent; pick N up front | Read cost of summing N grows painful → cache the aggregate / roll up | | HyperLogLog | Counts uniques in fixed tiny memory at huge scale | ~2% error; can't list members or do exact counts | Exact uniques or the member set is required → use a stored set | | Time-windowed buckets | Cheap rolling/rate counts; old data self-expires | More keys; window boundaries need care; cross-bucket reads sum many keys | You need an exact lifetime total → keep a separate lifetime counter | | Aggregate-on-read + cache | Cheap reads of a heavy-write count | Served total lags writes by the refresh interval | Reads must be fresh-to-the-write → read shards live (eat the N-sum) |

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-cr