“Design a distributed counter” sounds like homework until you write down what it actually is: a single number that millions of people want to add to at the same time. A TikTok video goes viral and its view count has to climb past a billion. An e-commerce flash drop decrements the same SKU’s stock from 10,000 to 0 while 2 million people hammer “buy.” A tweet’s like count, an ad’s impression count, a rate limiter’s request count - all the same shape. One logical integer, astronomically many writes converging on it, and a read path that wants the current value cheaply and often. The interviewer is not testing whether you can run count = count + 1. They are testing whether you understand that a single mutable number is the single worst thing to put under a write stampede, and whether you can pick the right consistency model for the specific counter in front of you.

That last part is the whole interview. There is no one distributed counter. A view count can be a few seconds stale and nobody is harmed - it wants raw throughput and tolerates eventual consistency. An inventory count that controls whether you oversell a product cannot be wrong even for a moment - it wants strong consistency and a hard floor at zero. If you reach for the same design for both, you either melt under load (strong consistency on views) or you oversell and refund angry customers (eventual consistency on stock). The entire design is choosing, per counter, where on that spectrum you sit, and building the machinery that makes the choice fast. Let me build both ends properly.

This overlaps with a flash-sale design, but a flash sale is one specific counter - a hard cap counting down to zero exactly once. This post is the general machine: counters that count up without bound at millions of increments per second (views, impressions) and counters that count down against a floor (inventory), and the single framework that covers both. The hard part shifts from “one campaign’s cap” to “pick consistency per counter and shard a hot number thousands of ways without losing increments.”

Functional Requirements (FR)

In scope:

  • Increment a counter. A client emits “+1” (or “+N”) against a named counter - views:{video_id}, stock:{sku}, likes:{post_id}. This is the overwhelming majority of traffic.
  • Decrement a counter. Symmetric: “-1” or “-N”, used for inventory, un-likes, cancellations. Some counters never go below a floor (stock >= 0).
  • Read a counter. Return the current value of a named counter. For view counts this can be approximate/stale; for inventory it must be exact at decision time.
  • Many counters, independently. The system holds billions of distinct counters (one per video, per SKU, per post), each with wildly different hotness. A handful are white-hot; the long tail is nearly idle.
  • Batch/bulk read. “Give me the view counts for these 50 videos in one call” for a feed render.
  • Two consistency tiers. The system must support an eventually consistent tier (high throughput, approximate, for views/likes/impressions) and a strongly consistent tier (exact, floored, for inventory/quota) under one API.

Explicitly out of scope (own the scope):

  • Analytics / time-series rollups. “Views per hour for the last 30 days” is a separate aggregation pipeline (a metrics system). We expose the current total, and optionally feed a stream to analytics, but windowed queries are not this system.
  • Per-event storage. We do not keep every individual “+1” forever as a row. That is an events/metrics problem. We keep counts.
  • Fraud / bot-view filtering depth. We mention dedup and basic abuse control; a full view-fraud ML pipeline is its own project.
  • Rich transactions across many counters. We support single-counter atomicity and a limited multi-counter conditional path for inventory, not general distributed transactions.
  • Exactly-once semantics as a blanket promise. We provide idempotent increments where needed (inventory) and accept at-least-once with dedup for the approximate tier. More on this below.

The one decision that drives everything: a counter is a shared mutable cell, and shared mutable state under concurrency is the enemy. The design is a set of techniques for not having a single shared cell - sharding it, making the writes commutative, batching them, and only reconciling to a single number when someone reads. The consistency tier decides how aggressively you are allowed to do that.

Non-Functional Requirements (NFR)

  • Scale: billions of counters. Aggregate write rate across the fleet of tens of millions of increments/sec at peak. A single hot counter (a viral video) can take 1M+ increments/sec by itself. Reads are heavier still - a feed render reads many counts; call it 10:1 read:write in aggregate.
  • Latency: increment ack in under ~10ms (often fire-and-forget, so the client does not even wait). Read in under ~20ms p99 for the approximate tier; the strong tier read-for-decision in under ~50ms.
  • Availability: 99.99% on the increment path. Losing a few increments on the approximate tier during an incident is acceptable; the counter is self-healing as new increments arrive. The strong tier must not lie.
  • Consistency: two explicit tiers.
    • Eventually consistent tier (views, likes, impressions): increments are commutative and associative; the value converges; reads may be seconds stale. No increment is lost (durability), but ordering and instantaneous accuracy are not promised.
    • Strongly consistent tier (inventory, quota, balance): the value is linearizable at decision time; a conditional decrement (stock >= 1) is atomic; the floor (>= 0) is never violated. Overselling is a correctness bug, not a UX blemish.
  • Durability: acknowledged increments survive node crashes. For the approximate tier we tolerate losing the last few hundred ms of un-flushed increments on a hard crash if replication is async; for the strong tier, an acknowledged decrement is durable before it is acknowledged.
  • Correctness invariants:
    • Approximate tier: displayed_value converges to sum(all increments) - sum(all decrements) as writes quiesce; monotonic where the counter is monotonic (views never go down).
    • Strong tier: value >= floor always; a decrement that would cross the floor is rejected atomically.

Back-of-the-Envelope Estimation (BoE)

Real numbers, with the arithmetic, because they dictate where the pressure is.

Write load (the whole problem):

Say the platform serves 2B video views/day across all videos.
  2,000,000,000 / 86,400 s = ~23,000 views/sec average.
Peak is ~5x average => ~115,000 increments/sec steady, fleet-wide, just for views.
Add likes, shares, impressions, inventory decrements => call it ~1,000,000 increments/sec fleet-wide at peak.

That average is boring. The danger is the distribution, not the mean:

One viral video during a live event: 1,000,000 views/sec onto ONE counter.
A single counter key at ~100,000 ops/sec (optimistic for a durable store) is 10x under water.
That ONE hot key, not the fleet total, is what breaks the naive design.

The skew is the story. Fleet-wide a million writes/sec is easy to spread. But those writes are not spread - they pile onto a few thousand hot counters while a billion counters sit idle. Any design that routes a counter to a single owner has a hot-key problem the moment one counter goes viral.

Read load:

Feed render reads ~20 counts (views + likes for each visible item).
200M daily active users, ~30 feed refreshes/day => 6B feed renders/day.
6B * 20 = 120B count reads/day = ~1.4M reads/sec average, ~7M/sec peak.

Reads dwarf writes and are extremely cache-friendly - the same hot counters are read over and over. A read-through cache with a short TTL absorbs nearly all of it.

Storage:

Per counter record ~ 64 bytes (counter_id, value, updated_at, a little metadata).
2,000,000,000 counters * 64 bytes = ~128 GB for the base values. Trivial, fits sharded in memory + disk.
Sharded counters (next): if hot counters get 256 sub-shards each but only ~10,000 counters are ever hot,
  10,000 * 256 * 64 bytes = ~164 MB of extra shard rows. Noise.

Storage is not the problem. This is not big data; it is a write-contention and consistency problem on a tiny amount of state.

Bandwidth:

Increment payload ~100 bytes. 1M/sec * 100 B = 100 MB/s ingest. Easy.
The pain is not bytes; it is lock/serialization contention on individual hot keys.

The takeaways: the fleet total is easy; the skew is brutal - a single counter can want 1M writes/sec. Reads are huge but cache-friendly. State is tiny. So the entire design budget goes into (1) never serializing a hot counter through one owner, (2) making increments commutative so they can be applied in any order on any node, and (3) picking strong vs eventual per counter so inventory is exact while views are cheap.

High-Level Design (HLD)

The architecture splits by consistency tier from the front door. Increments land on a stateless ingest tier that, for the approximate tier, buffers and batches them before touching storage, and for the strong tier routes them to the counter’s authoritative shard for an atomic conditional update. Both tiers persist to a sharded, replicated store. Reads come from a read-through cache that aggregates sub-shards into a single value.

   ┌──────────────┐    +1 view / -1 stock / read     ┌──────────────────────┐
   │   Clients     │ ───────────────────────────────▶ │   CDN / Edge          │
   │ (apps, web,   │                                   │ - caches read values  │
   │  services)    │ ◀─────────────────────────────── │   (short TTL)         │
   └──────────────┘            current value           └──────────┬───────────┘
                                                                  │
                                                       ┌──────────▼───────────┐
                                                       │  API Gateway / LB     │
                                                       │  authn, rate limit,   │
                                                       │  route by tier        │
                                                       └─────┬──────────┬──────┘
                              approximate tier (views)       │          │  strong tier (inventory)
                   ┌──────────────────────────────────▼──┐   │   ┌──────▼─────────────────────┐
                   │   Ingest / Aggregator (stateless)    │   │   │  Counter Service (strong)   │
                   │  - local in-memory delta buffer      │   │   │  - route to counter's shard │
                   │  - coalesce N increments into 1 add  │   │   │  - atomic conditional DECR  │
                   │  - flush every ~1s or every K ops    │   │   │    (value >= floor)         │
                   └───────────────┬──────────────────────┘   │   └──────┬──────────────────────┘
                                   │ batched "+K" per shard    │          │ atomic per-key op
                   ┌───────────────▼──────────────────────────▼──────────▼──────────────────────┐
                   │                Sharded, replicated counter store (per-key owner)             │
                   │   approximate: counters split into S sub-shards   strong: one owner per key   │
                   │   row = (counter_id, shard_id, value)             row = (counter_id, value)    │
                   └───────────────┬───────────────────────────────────────────────┬──────────────┘
                                   │ sum sub-shards on read                          │ exact read
                   ┌───────────────▼───────────────┐                  ┌──────────────▼──────────────┐
                   │  Read-through cache (Redis)     │                 │  Replication / WAL (durable  │
                   │  value = sum(sub-shards), TTL~1s │                │  log, quorum writes)         │
                   └────────────────────────────────┘                 └─────────────────────────────┘
                                   │ optional tap
                   ┌───────────────▼───────────────┐
                   │  Stream (Kafka) -> analytics    │  (time-series rollups, out of scope here)
                   └────────────────────────────────┘

Increment flow, approximate tier (views/likes) - the cheap, high-throughput path:

  1. Client fires “+1” for views:{video_id}. It is fire-and-forget; the client does not block on the true total.
  2. The increment lands on any stateless Ingest/Aggregator node (load-balanced, no stickiness needed). The node adds 1 into a local in-memory delta buffer keyed by counter.
  3. Every ~1 second (or every K increments), the node flushes the buffer: for each counter it has seen, it issues one ADD +K to a randomly chosen sub-shard of that counter in the store. A thousand “+1"s from one node become one “+1000” write. This coalescing is what turns 1M increments/sec into a manageable write rate at the store.
  4. The store applies value = value + K on that sub-shard - a commutative, order-independent add. Replication propagates it; the ack is fast because order does not matter.

Increment flow, strong tier (inventory) - the exact path:

  1. Client sends “-1” for stock:{sku} and waits for the result (did I get the item?).
  2. The Counter Service routes the request to the single authoritative shard that owns that key (consistent hashing on counter_id).
  3. That shard performs an atomic conditional decrement: UPDATE ... SET value = value - 1 WHERE counter_id = ? AND value >= 1. The write is quorum-committed (durable) before acking. If value < 1, it returns “out of stock” without decrementing. No coalescing, no sub-shards - correctness requires one owner and a real transaction per op.

Read flow:

  1. A read for views:{video_id} hits the read-through cache (Redis/edge) with a ~1s TTL. On a hit it returns immediately - this absorbs the 7M reads/sec.
  2. On a miss, the cache sums the sub-shards for that counter from the store (SELECT SUM(value) WHERE counter_id = ?, or a scatter-gather over S shard owners), stores the total with a short TTL, and returns it. The value is seconds-fresh, which is exactly the approximate contract.
  3. A strong-tier read (inventory at checkout) bypasses the cache and reads the authoritative shard directly for a linearizable value, because a decision is being made on it.

The key insight: the approximate tier never lets a hot counter serialize anywhere - increments buffer locally, coalesce, and land as commutative adds on random sub-shards, then get summed only when someone reads. The strong tier accepts serialization on one owner because inventory must be exact, and it keeps that owner survivable by giving each key its own small transaction rather than a global lock. One API, two machines, chosen per counter.

Component Deep Dive

The hard parts, naive-first then evolved: (1) the hot-key write problem and sharded counters, (2) strong vs eventual consistency and where each is mandatory, (3) commutative increments / CRDTs and why “+N” is special, and (4) durability and not losing or double-counting increments.

1. The hot key: one viral counter wants a million writes/sec

This is the heart of the design. A single counter under a write stampede.

Naive approach: one row, read-modify-write.

val = SELECT value FROM counters WHERE id = ?       -- read
UPDATE counters SET value = ? WHERE id = ?           -- write val+1 back

Where it breaks:

  • Lost updates (the classic race). Two increments both read value = 1000, both write back 1001. Two “+1"s, but the counter moved by one. Under a viral spike this does not misfire rarely - it drops a large fraction of increments. The view count silently runs low.
  • Even done atomically, the row is a serialization point. Fix the race with UPDATE ... SET value = value + 1 (atomic in the DB), and you still funnel every one of 1M increments/sec through a single row’s lock. A hot row does maybe a few thousand writes/sec before lock contention collapses throughput. You are 100x+ over. The whole shard melts, and it takes neighboring counters on the same node down with it.

First evolution: atomic add removes the lost-update race, not the hot key.

UPDATE counters SET value = value + 1 WHERE id = ?   -- atomic, no read needed

This is correct - no lost updates, the increment is commutative. But it does nothing for throughput: one row, one lock, one owner. Correct and still on fire. The problem was never correctness of the single add; it is that a single key cannot absorb the write rate.

Second evolution: split the hot counter into S sub-shards.

Instead of counters[video_id] = value, store counters[video_id][0..S-1], each an independent sub-counter, spread across different nodes. An increment picks a sub-shard at random (or round-robin) and does an atomic add there:

shard = random(0, S)                 -- spread the writes
ADD counters:{video_id}:{shard} += 1 -- atomic on a DIFFERENT key each time
read value = SUM over shard in 0..S-1 of counters:{video_id}:{shard}
  • Throughput multiplies by S. With S = 256 sub-shards across the cluster, a 1M writes/sec counter becomes ~4,000 writes/sec per sub-shard - comfortably inside a single key’s budget. The hot key is gone; it is now 256 warm keys on 256 different nodes.
  • Correctness holds because addition is commutative and associative. It does not matter which sub-shard a “+1” lands on or in what order the sub-shards are applied; sum(sub-shards) is the same total. This is why counters shard so cleanly and, say, a bank balance with per-transaction ordering rules does not.
  • The cost moves to the read. A read now sums S values instead of reading one. That is fine: reads are cache-fronted with a short TTL, so the scatter-gather-and-sum happens once per TTL per counter, not per read. The sum of 256 small integers is microseconds.

The adaptive catch: not every counter is hot. Sharding every one of 2B counters into 256 sub-shards is wasteful - 512B rows, and a read of a cold counter now scatter-gathers 256 nodes to find 255 zeros. The fix is adaptive sharding: counters start with S = 1 (a single key). A hotness detector (write rate over a sliding window at the ingest tier) promotes a counter to more sub-shards when it crosses a threshold, and the shard count is recorded in a small counter_meta table so reads know how many sub-shards to sum. Cold counters stay cheap; only the few thousand genuinely hot counters pay the sharding tax. State the trade-off: more sub-shards buys write throughput at the cost of read fan-out and storage, so you spend shards only where the write heat actually is.

2. Strong vs eventual consistency: picking per counter

This is the decision the interviewer is really probing. Same data structure, opposite requirements.

Naive approach: pick one model for everything.

Either make every counter strongly consistent (one owner, transactional conditional writes) or make every counter eventually consistent (sharded commutative adds, summed on read).

Where it breaks - both choices are wrong for half the counters:

  • Strong consistency on view counts. You force every “+1” view through one authoritative owner with a quorum-committed transaction. The viral video’s owner serializes 1M writes/sec through a consensus round-trip and dies. You have spent a linearizability guarantee that nobody needs - nobody is harmed if the view count is 1,000,240 vs 1,000,238 for a second. You paid the maximum price for a property with zero value here.
  • Eventual consistency on inventory. You shard stock:{sku} into sub-shards and let decrements land commutatively. Now two problems: (a) the floor is unenforceable - a sub-shard can go negative while another is positive, and “sum >= 0” is not the same as “never sold the 10,001st unit”; you oversell. (b) A read is stale, so two customers both see “3 left,” both buy, and you are now refunding one. Eventual consistency on a resource that gates a real-world commitment is a correctness bug.

The answer: classify the counter, route to the matching engine.

PropertyApproximate tier (eventual)Strong tier (linearizable)
Exampleviews, likes, impressions, reactionsinventory, quota, seats, balance
StorageS sub-shards, commutative addssingle owner per key
Write pathbuffer + coalesce + random sub-shardatomic conditional txn on owner
Floor/capnone (monotonic or free)enforced atomically (value >= floor)
Readsum sub-shards, cache ~1s, stale okread owner directly, linearizable
Throughputmillions/sec per counterbounded by one owner (~thousands/sec)
On conflictconverge, never rejectreject (out of stock)
  • Approximate tier uses the sharded commutative design from part 1. It is fast precisely because it gives up ordering and instantaneous accuracy - properties the counter does not need. It is eventually consistent: increments are never lost (durable), and the value converges to the true sum as writes quiesce.
  • Strong tier keeps one authoritative shard per key and does a quorum-committed conditional update per operation: SET value = value - 1 WHERE value >= 1. The condition and the write are one atomic, linearizable step, so the floor is never crossed and no two buyers get the last unit. It is deliberately not sharded into sub-shards, because you cannot enforce a global floor across independently-updated sub-shards without a global lock, and a global lock is just a slow single owner. One owner per key, sized so that no single inventory key needs millions of writes/sec (inventory decrements are bounded by actual stock, which is finite and usually modest).

The interview answer in one sentence: shard and go eventual when the counter only needs to converge; keep one owner and go strong when the counter gates a real commitment with a floor. The flash-sale “6M cap” is the strong tier with a hard cap; a viral view count is the approximate tier. Same system, two engines.

The uncomfortable middle: a hot inventory counter. What if one SKU is both inventory (needs a floor) and takes 1M decrements/sec (a flash drop)? You cannot shard it freely (floor) and cannot single-own it (throughput). The resolution is reservation sharding: pre-allocate the finite stock across sub-shards as fixed budgets (10,000 units split into 100 shards of 100 each), decrement locally and atomically within a shard with its own local floor, and retry on another shard when one empties. The global floor holds because sum(shard budgets) = total stock and each shard floors at its own zero. This is exactly the flash-sale technique, and it is the bridge between the two tiers - you get throughput by sharding a bounded resource whose shard caps sum to the true cap. It costs tail fairness (some shards drain first), which is acceptable for a sale and unacceptable for, say, a bank balance.

3. Commutative increments and CRDTs: why “+N” is special

The reason counters shard so well is algebraic, and naming it shows depth.

Naive framing: a counter is a value you overwrite. Treat increment as “write the new value.” Then concurrent writers conflict - last-writer-wins drops increments, and you need coordination (locks, consensus) to serialize every write. That is the strong tier, and it is slow.

Where it breaks: overwrite semantics force a total order on operations, which forces coordination, which forces a single serialization point. You have recreated the hot key by definition.

The answer: model the counter as a set of commutative operations, not a value.

An increment is +N. Addition is commutative (a + b = b + a) and associative ((a+b)+c = a+(b+c)). That means concurrent increments on different replicas never conflict - you can apply them in any order, on any node, and merge by summing. This is a CRDT - a Conflict-free Replicated Data Type - specifically a PN-Counter (Positive-Negative counter):

PN-Counter for counter C:
  P = vector of per-replica increment totals  (monotonically increasing)
  N = vector of per-replica decrement totals  (monotonically increasing)
  value(C) = sum(P) - sum(N)
  merge(stateA, stateB) = element-wise max on P and on N      -- idempotent, commutative
  • Each replica owns one slot in P and one in N and only ever increases its own slots. Merging two replicas’ states takes the element-wise max, which is commutative, associative, and idempotent - so merges can happen in any order, repeat harmlessly, and the value converges. This is the formal backbone of the approximate tier: the S sub-shards are effectively the per-replica slots, and sum(sub-shards) is value(C).
  • Idempotent merge is the superpower. Because applying a replica’s state twice changes nothing (max is idempotent), the replication protocol can be dumb and at-least-once - re-delivering a state is safe. This is why the approximate tier survives retries and network flakiness without double-counting at the state level.
  • The limit CRDTs do not cross. A PN-Counter has no way to enforce “never below zero” - you cannot look at sum(P) - sum(N) on one replica and know whether another replica already decremented past the floor without coordinating. That is the mathematical reason inventory cannot be a plain CRDT and must use the strong tier (or reservation-sharded budgets with per-shard floors). The floor requires a point of agreement; commutativity gives it up.

Naming CRDT/PN-Counter and then naming its limit (no global invariant like a floor) is the move that separates “I memorized sharded counters” from “I understand why they shard.”

4. Durability and double-counting: at-least-once vs exactly-once

An acknowledged increment must not vanish, and a retried increment must not be counted twice where that matters.

Naive approach: buffer increments in memory, flush later, ack immediately.

The ingest node adds “+1” to its local buffer and tells the client “ok” before the increment is durable.

Where it breaks:

  • Crash loses the buffer. The node holds a buffered “+500” and crashes before flushing. 500 views evaporate. For a view count this is a tolerable blip (self-healing as new views arrive), but you must decide it is tolerable, not discover it.
  • Double-counting on retry. The client times out waiting for an ack and retries “+1.” Now the buffer has “+2” for one real view. At the value level, PN-Counter state-merge is idempotent, but an operation-based “+1 again” is not - naive op replay inflates the count. For views, mild inflation during retries is acceptable; for inventory or billing it is not.

The answer: match the delivery guarantee to the tier.

  • Approximate tier: at-least-once + bounded loss, self-healing. Ack after the increment is in a replicated buffer (the ingest node’s buffer is itself replicated, or increments are first appended to a short Kafka log), so a single node crash loses nothing; only a correlated failure loses the last sub-second window, which the monotonic counter absorbs. Accept that view counts can over- or under-count by a rounding-error fraction during incidents. This is the honest contract: eventually consistent and approximately exact, not perfect.
  • Approximate tier, when you want it tighter: dedup with idempotency keys. For likes (where “did my like register” is visible to the user), each increment carries an idempotency key (user_id:post_id for a like is naturally unique). The ingest tier keeps a short-TTL dedup set; a repeat key is dropped. This turns like-counting effectively exactly-once at the semantic level (one user, one like) without coordination, because the key is the dedup.
  • Strong tier: synchronous durability + idempotent operations. The conditional decrement is quorum-committed to a WAL before acking - no buffering, no “ok” before durable. Retries carry an idempotency key (order_id), and the owner records applied keys so a replayed decrement is a no-op that returns the original result. This gives true exactly-once for the operations that gate money and stock. It is slower (a consensus/quorum round-trip per op), which is exactly why you reserve it for counters that need it.

The principle: durability and exactly-once are not free, so buy them only for the counters that break without them. Views buy cheap at-least-once with self-healing; inventory buys expensive exactly-once with quorum writes and idempotency keys. Spending the strong guarantee on views is how you melt the system; spending the cheap guarantee on inventory is how you oversell.

API Design & Data Schema

The API is small and declares the tier explicitly so the system routes correctly.

REST / gRPC endpoints

POST /api/v1/counters/{id}/increment
  headers: Idempotency-Key: <optional, required for strong tier>
  body: { delta: 1, tier: "approx" | "strong", floor: 0 }
  -> 200 { accepted: true }                         (approx: fire-and-forget ack)
  -> 200 { value: 42, accepted: true }              (strong: new exact value)
  -> 409 { accepted: false, reason: "floor_reached" }   (strong: would cross floor)

GET /api/v1/counters/{id}
  ?consistency=approx|strong
  -> 200 { value: 1204387, as_of: "2026-11-13T00:00:01Z", approximate: true }   (approx)
  -> 200 { value: 3, approximate: false }                                        (strong, linearizable)

POST /api/v1/counters/batchGet
  body: { ids: ["views:v1","likes:v1", ...up to 100], consistency: "approx" }
  -> 200 { "views:v1": 1204387, "likes:v1": 88123, ... }   (one round trip for a feed)

POST /api/v1/counters/{id}/decrement        (sugar for increment with delta negative)
  body: { delta: 1, tier: "strong", floor: 0, idempotency_key: "<order_id>" }
  -> 200 { value: 2 } | 409 { reason: "floor_reached" }

The increment on the strong tier is idempotent via the idempotency key, so retries never double-decrement stock. The approximate tier increment is fire-and-forget by default - the client does not wait for or trust the returned value. batchGet is the feed-render workhorse and is served almost entirely from cache.

Data stores - the right tool per job

1. Approximate counters - sharded commutative store (NoSQL/KV or Redis). The access pattern is ADD key += N (commutative) and SUM over sub-shards on read. No joins, no transactions, extreme write throughput, tiny records. A wide-column store (Cassandra, with its counter columns, or ScyllaDB) or Redis cluster fits - both do atomic adds and shard by key.

Table: counter_shards        (NoSQL, partition key = counter_id, clustering = shard_id)
  counter_id     (PARTITION KEY)     -- e.g. "views:video_42"
  shard_id       (CLUSTERING KEY)    -- 0 .. S-1
  value          COUNTER / BIGINT    -- atomic add; commutative
  updated_at     TIMESTAMP
  Read value: SELECT SUM(value) FROM counter_shards WHERE counter_id = ?
  Replication: factor 3, async / tunable (ONE for writes, QUORUM for strong reads)

Table: counter_meta          (how many sub-shards a counter has)
  counter_id     (PARTITION KEY)
  shard_count    INT          -- adaptive: 1 for cold, up to 256+ for white-hot
  tier           ENUM(approx, strong)
  floor          BIGINT NULL  -- strong tier only
  updated_at     TIMESTAMP

Chosen because the write is a native atomic counter add and sharding by counter_id spreads the fleet; sub-shards (clustering key) spread a single hot counter across nodes. We do not need SQL’s transactions on this tier - we need commutative throughput.

2. Strong counters - single-owner store with real transactions (SQL or a consensus KV). The access pattern is a conditional atomic decrement with a floor and a durable quorum commit. This wants ACID on a single key: Postgres (row-level atomic UPDATE ... WHERE value >= 1), DynamoDB with a conditional write, or an etcd/consensus-backed key.

Table: strong_counters       (SQL or conditional-write KV)
  counter_id     PRIMARY KEY          -- e.g. "stock:sku_9"
  value          BIGINT NOT NULL      -- exact
  floor          BIGINT NOT NULL DEFAULT 0
  version        BIGINT               -- optimistic concurrency / audit
  updated_at     TIMESTAMP
  Decrement: UPDATE strong_counters SET value = value - :n, version = version + 1
             WHERE counter_id = :id AND value - :n >= floor   -- atomic conditional
             -- rows_affected == 1 -> success; 0 -> floor_reached

Table: applied_ops           (idempotency ledger for exactly-once decrements)
  idempotency_key  PRIMARY KEY        -- e.g. order_id
  counter_id       STRING
  result_value     BIGINT             -- returned on replay so retries are no-ops
  applied_at       TIMESTAMP

Why two stores: the two tiers have opposite needs. The approximate counter is write-dominated, commutative, and tolerant of staleness - a sharded KV with native atomic adds owns it and scales horizontally. The strong counter needs a linearizable conditional write and a durable floor - a transactional/consensus store owns it and is deliberately single-owner per key. Matching the store to the consistency requirement is the whole schema decision, and it is why “one database for all counters” is the wrong answer.

Durability detail (the subtle bit). On the approximate tier, the ingest buffer is a durability risk: buffered increments not yet flushed are lost on crash. The fix is to either (a) replicate the buffer, or (b) append raw increments to a short Kafka log before acking and have the aggregator consume-and-coalesce from the log, so a crash just reprocesses the log (at-least-once, idempotent-at-value-level). The strong tier never buffers - it quorum-commits each conditional write before acking, trading throughput for the guarantee.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each:

1. The hot key (breaks first, always). One viral counter wants 1M+ writes/sec; a single row/key tops out at a few thousand. Fix: split the counter into S sub-shards across nodes, commutative atomic adds on a random sub-shard, sum on read. Adaptive: only hot counters get many sub-shards, so cold counters stay cheap. This is the single most important technique in the whole design.

2. Lost updates under concurrency. Read-modify-write drops increments. Fix: atomic add (value = value + N), never read-then-write. Addition is commutative, so concurrent adds compose correctly. Removes the race independent of sharding.

3. Choosing the wrong consistency. Strong on views melts; eventual on inventory oversells. Fix: two tiers routed per counter - sharded commutative + eventual for views/likes, single-owner conditional + linearizable for inventory/quota. The floor-bearing counters get a real transaction; the free-running counters get throughput.

4. Read storm on hot counters. 7M reads/sec, concentrated on the same hot counters. Fix: read-through cache with a short TTL (~1s) in front of the sum-of-sub-shards. The scatter-gather-and-sum happens once per TTL per counter; everything else is a cache hit at the edge. Reads are the easy part precisely because they are cacheable.

5. Read fan-out on over-sharded counters. Summing 256 sub-shards for a cold counter is wasteful. Fix: adaptive shard counts stored in counter_meta; reads sum only the counter’s actual shard_count. A cold counter reads one key; only hot counters pay the fan-out, and their reads are cache-amortized anyway.

6. Flush/batch amplification at the ingest tier. Millions of tiny “+1"s hitting the store directly. Fix: local buffer + coalesce + periodic flush turns N increments from one node into one +N write per counter per flush interval. This is the write-side analog of caching and is what makes fleet-wide write rates land softly on the store.

7. Buffer loss on crash (durability). Un-flushed increments vanish. Fix: replicate the buffer or front it with a Kafka log; approximate tier tolerates bounded sub-second loss and self-heals, strong tier never buffers and quorum-commits before ack.

8. Double-counting on retries. At-least-once delivery inflates counts. Fix: idempotency keys - natural keys for likes (user:post), explicit keys (order_id) for strong decrements with an applied_ops ledger. State-based CRDT merges are idempotent at the value level; op-based retries need the key.

9. The hot-and-floored counter (flash-drop inventory). Needs both throughput and a global floor. Fix: reservation sharding - split the finite stock into fixed per-shard budgets whose caps sum to the true total, decrement locally with a per-shard floor, retry on empty. Buys throughput by sharding a bounded resource; costs tail fairness.

10. Rebalancing and resharding hot counters live. A counter goes viral mid-flight and needs more sub-shards. Fix: promote shard_count online - new increments start landing on the higher range; reads sum the new count. Because adds are commutative and sub-shards are independent, adding sub-shards is safe with no data movement; old sub-shards just keep their accumulated value and are included in the sum.

11. Single points of failure. Fix: approximate store replicated (factor 3), stateless ingest autoscaled, cache replicated; strong store replicated with consensus/quorum and automatic failover; the system fails safe per tier - approximate reads may serve a slightly stale value under partition (acceptable), strong reads require quorum and refuse rather than return a value they cannot linearize (never oversell).

Wrap-Up

The trade-offs that define this design:

  • Two tiers, not one model. The core move is refusing to pick a single consistency model. Views/likes/impressions run eventually consistent on a sharded, commutative, cache-fronted engine for raw throughput; inventory/quota/balance run linearizable on a single-owner conditional-write engine for an exact floor. One API routes per counter.
  • Shard the hot key, commutatively. A counter is a sum of independent sub-shards because addition is commutative and associative (a PN-Counter CRDT), so a 1M writes/sec counter becomes 256 warm keys with no coordination. Adaptive sharding spends shards only where the write heat is, keeping the billion cold counters cheap.
  • Coalesce writes, cache reads. Local buffering turns N increments into one +N flush; a short-TTL read-through cache turns millions of reads into one sum-per-interval. Both sides are absorbed before they reach the store.
  • Exactly-once only where it pays. Views buy cheap at-least-once with self-healing and natural-key dedup; inventory buys expensive quorum-committed exactly-once with idempotency keys and an applied-ops ledger. Spending the strong guarantee everywhere is how you melt; spending the cheap one on inventory is how you oversell.
  • The CRDT limit is the design boundary. Commutative counters cannot enforce a global floor without coordination, which is the mathematical reason inventory leaves the approximate tier. The hot-and-floored case is bridged by reservation sharding - sharding a bounded resource whose shard caps sum to the true cap - at the cost of tail fairness.

One-line summary: a distributed counter where approximate counters (views, likes) are split into adaptively-many commutative sub-shards, incremented via buffered-and-coalesced atomic adds, and summed behind a short-TTL cache for throughput and eventual consistency, while strong counters (inventory, quota) keep a single authoritative owner doing quorum-committed conditional decrements with an atomic floor and idempotency keys for linearizable exactness - one API that picks strong versus eventual per counter so view counts never melt the system and stock counts never oversell.