beginnerintermediateadvanced

System Design · Section 94

Distributed Counters

Level: intermediate
Read: 15 min
Concepts: 2

A counter stores the smallest possible amount of data and becomes one of the most expensive things in a system under load, because its cost is set by contention rather than by size. Incrementing means updating one specific row, and a database serialises concurrent updates to a row behind a lock held until commit — so the maximum increment rate is one divided by that hold time, and no number of application servers, connections or cores raises it. Lock hold time is the whole transaction, which means an unrelated slow statement in the same block lowers the counter's ceiling directly, and the cheapest available fix is often shortening the transaction rather than restructuring anything. The approach to the ceiling is non-linear: latency is negligible at 40% utilisation, noticeable at 80%, and unbounded once demand exceeds capacity, so a normal growth month can move a healthy counter into failure. The failure also does not stay local, because blocked transactions hold connections and an exhausted pool degrades queries that never touched the counter. Four techniques remove the serialisation point, each charging in a different currency. A sharded counter splits one logical value across N rows, keeping the total exact and paying in read cost — sized from the measured write rate, and pointless on a read-heavy counter, where the correct answer is a cache in front of one row. Batching buffers increments and flushes periodically, giving the largest reduction and paying in freshness and in whatever is buffered when a process dies. An approximate counter answers cardinality questions in fixed memory with a bounded error, and must be displayed at the precision it actually offers, since a verbatim estimate that occasionally moves downward reads as a defect. Event aggregation appends each event and computes counts on a schedule, making the write path contention-free and the count derived — which means a counting bug is fixed by recomputing rather than by repairing a number that has drifted. Which currency you can spend is decided by what the count is for.

System Design overview

System Design for Beginners CoursefreeCodeCamp.org

What is true here

A counter's ceiling is one divided by the lock hold time, and horizontal scaling does not raise it — the contended resource is one row.
Lock hold time is the entire transaction, so unrelated slow work in the same block lowers the counter's throughput.
Blocked transactions hold connections, so a hot counter can degrade operations that have nothing to do with it.
Sharding is exact and pays in read cost; batching pays in freshness and crash loss; approximation pays in exactness; aggregation pays in liveness.
What the number is for — a balance, an invoice total, a view count — decides which of those costs is acceptable.

What you will be able to do

Compute a counter's throughput ceiling from its transaction's lock hold time and compare it to peak demand
Explain why load tests with uniformly distributed keys miss per-key contention entirely
Choose between sharding, batching, approximation and aggregation from the counter's actual requirement
Size a sharded counter, and recognise when sharding would make a read-heavy counter worse

Why counters bottleneck

A single row, a lock held until commit, a hard ceiling and a non-linear approach to it.

Why a single counter row bottlenecks under write volume

coreintermediate

A counter looks like the cheapest thing in a database and becomes one of the most expensive under load, for a reason that has nothing to do with how much data it holds. Incrementing a counter means updating one specific row, and a database serialises concurrent updates to the same row: each transaction takes a row lock, applies its change, and holds that lock until it commits. So the maximum increment rate for one row is one divided by the time each transaction holds the lock — if a transaction holds it for two milliseconds, that row accepts about five hundred increments per second no matter how many application servers, connections or CPU cores you add. Everything else waits, and the waiting is what makes it worse: contending transactions occupy connections while blocked, so a hot counter can exhaust a connection pool and slow down operations that have nothing to do with it. The effect is invisible at low volume and non-linear at high volume, because queueing delay rises sharply as arrival rate approaches service rate. The counter is also usually not the point of the transaction — it is a view counter updated alongside a page load, or a like count updated alongside an insert — so the whole transaction's duration, including everything else it does, is what determines how long the lock is held. That is why the first fix is often not sharding the counter but shortening the transaction that touches it, and why the real fixes in the next concept all work by removing the requirement that every increment reach the same row.

Think of it as

One turnstile at a stadium. It does not matter how many people are outside or how many staff you hire; the throughput is set by how long each person takes to pass through it. Add a hundred more people and nothing speeds up — the queue just gets longer, and the queue itself starts causing problems, blocking the road, filling the concourse. A hot counter row is that turnstile, and the transaction time is how long each person takes.

sql

-- every one of these serialises on one row
BEGIN;
  UPDATE posts SET view_count = view_count + 1
   WHERE id = $1;          -- row lock taken here
  INSERT INTO view_log ...  -- lock still held
  UPDATE user_stats ...     -- lock STILL held
COMMIT;                     -- released only now

-- the counter's ceiling is set by everything
-- else in this transaction, not by the UPDATE

What we're doing: Watch a counter that is fine at 200 increments per second stop working at 600.

counter-ceiling.txttext

Transaction holding the row lock: 2 ms
Theoretical ceiling: 1 / 0.002 = 500 increments/s

200/s   utilisation 40%
        added latency: negligible
        Everything looks fine. This is the
        number the feature launched at.

400/s   utilisation 80%
        added latency: noticeable, ~8 ms
        Still "fine" on a dashboard of averages.

480/s   utilisation 96%
        added latency: ~48 ms and climbing
        Queue length grows faster than arrivals.

600/s   demand exceeds the ceiling
        the queue never drains; waits grow
        without bound until requests time out.
        Connections are consumed by blocked
        transactions, so unrelated queries start
        failing too.

Traffic tripled. Latency went from negligible to
unbounded. Nothing about the code changed.

8: Forty percent utilisation of a serialised resource is comfortable, which is why the feature ships and looks healthy for months. Nothing in the design says how close to the ceiling it is running.
16: This is the shape that makes contention dangerous. Between 80% and 96% utilisation the added latency grows roughly sixfold for a 20% traffic increase, so a normal growth month moves the system from fine to failing.
24: The blast radius is the important part: the failure does not stay inside the counter. Blocked transactions hold connections, and once the pool is exhausted every query in the application competes for what is left.

Why this works: The point is that a hot counter has a hard ceiling that no amount of horizontal scaling raises, and that the approach to that ceiling is non-linear rather than gradual. Knowing the transaction's lock hold time gives you the number, which turns "is this counter a risk" from a guess into arithmetic you can do before it becomes an incident.

Incrementing the counter at the start of a long transaction

Wrong

python

with db.transaction():
    db.execute("UPDATE posts SET view_count = "
               "view_count + 1 WHERE id = %s", pid)
    render_and_store_analytics(...)   # 40 ms
    update_recommendations(...)       # 25 ms
# the row lock is held for ~65 ms, so the
# counter's ceiling is about 15 increments/s

Better

python

render_and_store_analytics(...)
update_recommendations(...)
with db.transaction():                # separate,
    db.execute("UPDATE posts SET view_count = "  # and
               "view_count + 1 WHERE id = %s", pid)
# lock held for the length of one statement

What you see: A counter on a popular row starts timing out, and profiling shows the `UPDATE` itself is fast. The slow part is waiting for the lock, held by other transactions doing entirely unrelated work in the same block.

Why: A row lock is held until commit, so the counter inherits the duration of everything else in its transaction. Moving the increment into its own short transaction — or out of the request path altogether — raises the ceiling by the ratio of the transaction lengths, which is often more than an order of magnitude for a one-line change.

Every increment funnels through one row lock

App server 1

App server 2

App server 3

Row lock on posts.id = 9812

one holder at a time

view_count

ceiling = 1 / lock hold time

Everyone else waits

holding connections while blocked

Why the usual scaling levers do nothing here

Why the usual scaling levers do nothing here
Lever	Effect on a hot counter	Why
More application servers	None	They all queue on the same row lock
A larger connection pool	Worse	More connections wait on the same lock, and the pool is consumed by blocked work
A bigger database instance	Marginal	Faster commits shorten the lock slightly; the serialisation remains
Read replicas	None	The contention is on writes
Shortening the transaction	Real	Throughput is the inverse of lock hold time
Not writing to one row (next concept)	Structural	Removes the serialisation point entirely

Remember: A counter's throughput ceiling is one divided by how long the transaction holds its row lock, and no amount of horizontal scaling raises it, because the contended resource is one row rather than compute. Lock hold time is the whole transaction, so an unrelated slow statement in the same block lowers the ceiling directly. The approach to that ceiling is non-linear, and blocked transactions consume connections, so a hot counter degrades operations that have nothing to do with it. Shortening the transaction is the cheap fix; removing the single-row requirement is the real one.

The four ways out

Sharding, batching, approximation and event aggregation — and the different cost each one charges.

Sharded counters, batching, approximation and event aggregation

coreintermediate

Four techniques remove the single-row bottleneck, and each one gives up something different. A sharded counter replaces one row with N rows for the same logical value: each writer picks a shard at random and increments that, and a read sums all N. Write contention drops by roughly a factor of N, reads get more expensive, and the count stays exact. Batching accumulates increments in memory or in a fast store and flushes them periodically as one larger update: a thousand increments become one write, which is a hundredfold or thousandfold reduction in contention, at the cost of losing whatever was buffered if the process dies and of the displayed count lagging the flush interval. An approximate counter gives up exactness on purpose: a probabilistic structure such as HyperLogLog answers "how many distinct users viewed this" using a small fixed amount of memory with a bounded error, which is the right trade when the number is displayed rounded anyway. And event aggregation stops treating the count as a value to update at all — each event is appended to a log or stream, and a job computes counts from it on a schedule, which makes the write path an append with no contention and makes the count a derived, recomputable number rather than a stored one. The question that picks between them is what the count is for: an account balance must be exact and synchronous, a view count on an article can be approximate and minutes old, and treating those two the same is what produces either a wrong balance or an unnecessary bottleneck.

Think of it as

Four ways to stop the queue at the one turnstile. Open ten turnstiles and add up the tallies at the end (sharding). Let each entrance count on paper and phone the total in every minute (batching). Estimate the crowd from a sample rather than counting heads (approximation). Or stop counting at the door entirely, keep the ticket stubs, and count them later (event aggregation). Which one is right depends entirely on what the number will be used for, and that is a product question before it is an engineering one.

sql

-- sharded counter: N rows for one logical value
CREATE TABLE view_counts (
  post_id  uuid NOT NULL,
  shard    int  NOT NULL,        -- 0 .. N-1
  count    bigint NOT NULL DEFAULT 0,
  PRIMARY KEY (post_id, shard)
);

-- write: pick a shard at random, contend with
-- roughly 1/N of the writers
UPDATE view_counts SET count = count + 1
 WHERE post_id = $1 AND shard = floor(random() * 16);

-- read: sum the shards
SELECT sum(count) FROM view_counts WHERE post_id = $1;

What we're doing: Apply each technique to the same 600-per-second view counter and compare what it costs.

four-fixes.txttext

Problem: 600 increments/s on one row whose
ceiling is 500/s.

1. SHARDED, N = 16
   per-shard write rate: ~38/s, well under 500
   read: SELECT sum(count) over 16 rows
   exactness: exact
   cost: reads are 16 row lookups instead of 1

2. BATCHED, flush every 2 seconds
   in-process counter; one UPDATE per post per
   flush
   write rate on the row: 0.5/s
   exactness: exact unless a process dies
   holding a buffer
   cost: the count is up to 2 seconds stale

3. APPROXIMATE (distinct viewers)
   HyperLogLog per post, fixed ~12 KB
   write: one add, no contention
   exactness: bounded error (a fraction of a
   percent at typical precision)
   cost: cannot answer "exactly how many", and
   cannot be decremented

4. EVENT AGGREGATION
   append {post_id, ts, viewer} to a stream
   a job computes counts every minute
   write path: append-only, contention-free
   exactness: exact, one minute behind
   cost: a pipeline to run, and counts are not
   live

For a view counter, 2 or 4. For a like count
that must be exact and near-live, 1. For "how
many distinct people", 3.

8: Sixteen shards turns a 600/s problem into sixteen 38/s problems, each far below the ceiling. The technique is exact, which is why it stays available for counts where approximation is not acceptable.
17: Two seconds of staleness is invisible on a view counter and unacceptable on a balance. Batching is the highest-leverage technique here precisely because the requirement is weak — a thousandfold contention reduction bought with a delay nobody perceives.
30: Event aggregation gives up liveness and gets back something the others do not offer: the count is derived, so a counting bug is fixed by recomputing from the log rather than by trying to repair a drifted number nobody can reconcile.

Why this works: All four solve the same contention problem and are not interchangeable, because they charge for it in different currencies — read cost, durability, exactness, and freshness. Naming which of those four the specific counter can afford to spend is the whole decision, and it is answerable from the product requirement rather than from the database.

Sharding a counter that is read far more than it is written

Wrong

sql

-- 64 shards on an article's view count
-- writes: 4/s. reads: 30,000/s.
SELECT sum(count) FROM view_counts
 WHERE post_id = $1;    -- 64 rows read, 30,000
                        -- times per second

Better

text

-- The write rate was never the problem.
-- One row, plus a cached read:
--   GET view_count:{post_id}  (cache, 60s TTL)
-- Sharding solved a bottleneck that did not
-- exist and created one that did.

What you see: Database load rises sharply after introducing sharded counters, and the increase is entirely in reads. The counter's write latency, which was fine before, is still fine — nothing was gained and read volume was multiplied by the shard count.

Why: Sharding trades read cost for write throughput, which is only a good trade when writes are the constraint. Measuring the actual per-key write rate before sharding avoids applying a write-side fix to a read-side workload — where the correct answer is a cache in front of one row.

Four ways to remove the serialisation point

Sharded

N rows per counter

read = sum(N)

Batched

huge contention drop

buffer lost on crash

Approximate

cardinality questions

display it rounded

Aggregated

writes never contend

recomputable from the log

Four techniques and what each gives up

Four techniques and what each gives up
Technique	Contention reduction	Gives up	Best for
Sharded counter	~N times	Cheap reads	Exact counts with high write rates and low read rates
Batching	Very high	Buffered increments on a crash; freshness	High-volume counts where small loss is acceptable
Approximate counter	Very high	Exactness, and the ability to decrement	Cardinality questions displayed rounded
Event aggregation	Total — writes never contend	Real-time freshness	Anything that can be recomputed and does not need to be live

Match the technique to what the number is for

Match the technique to what the number is for
Counter	Requirement	Technique
Account balance	Exact, synchronous, must never drift	None of these — keep the transactional write
Remaining stock	Exact, synchronous, must not go negative	Conditional update with a constraint
Article view count	Approximate, minutes stale is fine	Batching or event aggregation
Distinct visitors today	Approximate cardinality, displayed rounded	Approximate counter
Likes on a viral post	Exact eventually, high write rate	Sharded counter, or aggregation
API usage for billing	Exact eventually, auditable	Event aggregation — the log is the evidence

Remember: Four ways out, each charging in a different currency. Sharding stays exact and pays in read cost — size N from the measured write rate, and do not shard a read-heavy counter. Batching pays in freshness and in whatever is buffered when a process dies, and gives the largest reduction. Approximate counters pay in exactness for cardinality questions, and must be displayed at the precision they actually offer. Event aggregation pays in liveness and returns a count that is derived and recomputable. What the number is for decides which currency you can spend.