Distributed Counter — Sharded with Async Aggregation

The sharded counter architecture is the industry-standard solution to the hot-key write bottleneck that makes the naive single-row approach unworkable. It is the pattern used by YouTube for view counts, Instagram for like counts, and Twitter for engagement metrics. The core insight is that a single counter does not need to be a single physical storage location — it can be split into N independent sub-counters that are summed on read.

The hot-key problem is severe. A viral video receiving 1M views per second means 1M writes per second to a single counter key. A single Redis key can sustain ~1M operations per second on a single core, but the key may be hashed to a single slot in a Redis cluster, limiting throughput to one core. A single PostgreSQL row can sustain only ~2K-5K updates per second. The sharded counter pattern solves this by distributing writes across N sub-counter keys, each potentially on a different Redis node or hash slot.

The architecture splits each logical counter into 16 sub-counters in Redis: counter:{key}:0 through counter:{key}:15. When an increment arrives, CounterService hashes (item_id, random_shard_index) to select one of the 16 shards and performs an atomic INCR. Each shard sees at most 1/16th of the total write traffic — 62,500 ops/sec instead of 1M. This is well within the capacity of a single Redis core (~1M ops/sec). The write path is fast (~2ms Redis INCR) and scales linearly with the number of shards.

The trade-off is read complexity. The current count is the sum of all 16 shards, requiring an MGET of 16 keys. To avoid this per-read overhead, an AggregatorWorker periodically (every 5 seconds) sums all shards and writes the total to a MasterCache key. Read requests fetch this single master key — fast and simple, but stale by up to 5 seconds. For a video at 100K views/sec, the displayed count might be 500K behind reality, but '12.3M views' vs '12.8M views' is indistinguishable to users.

Durability comes from a separate PersistWorker that batches aggregated counts and writes to CounterDB (DynamoDB) every 30 seconds. At 1M increments/sec, per-increment DB writes would require 1M write IOPS — prohibitively expensive. Batching reduces this to ~33K writes/sec (1M active keys / 30s), a 30x cost reduction.

The architecture uses Kafka (AggStream) as a buffer between the hot write path and the async aggregation/persistence path. During viral events, Kafka absorbs burst traffic (5M message buffer) while workers process at a sustainable rate. CounterService publishes increment events to Kafka as fire-and-forget — the user response does not wait for aggregation or persistence.

This pattern appears in virtually every system design interview for counter-related questions. Interviewers expect candidates to identify the single-row bottleneck, propose sharding as the solution, reason about the approximate-vs-exact read trade-off, and explain why async aggregation is necessary for read performance. The CRDT variant extends this to multi-region deployment with conflict-free convergence.

The sharded counter system uses 10 components organized into a write-optimized pipeline with async aggregation and persistence. The components are: CounterClient, ApiGateway, MainLB, CounterService (write path), ReadService (read path), ShardCache (Redis sub-counters), MasterCache (Redis aggregated totals), AggStream (Kafka), AggregatorWorker, PersistWorker, and CounterDB (DynamoDB).

All traffic enters through the ApiGateway, which performs JWT authentication (~3ms), enforces a 1.2M RPS rate limit, and routes all /api/v1/ traffic to MainLB. The MainLB distributes traffic between CounterService (increments) and ReadService (count reads) using round-robin.

The write path is optimized for maximum throughput with minimum latency. CounterService receives POST /api/v1/counters/{key}/increment, hashes (item_id, random_shard_index) to select one of 16 sub-counter shards, and performs an atomic INCR on ShardCache (Redis). The write completes in ~2ms. No database is touched on the hot write path. CounterService then publishes an increment event to AggStream (Kafka) as fire-and-forget for downstream aggregation.

The aggregation pipeline runs continuously in the background. AggregatorWorker consumes from AggStream and, every 5 seconds per active key, reads all 16 sub-counter shards from ShardCache via MGET, computes the sum, and writes the aggregated total to MasterCache. This master count is the approximate value — stale by at most one aggregation interval (5 seconds). PersistWorker batches aggregated counts and writes to CounterDB every 30 seconds for durable storage.

The read path has two modes. Approximate reads (GET /api/v1/counters/{key}) fetch the pre-aggregated value from MasterCache in ~2ms — a single Redis GET. This is used by 95% of read traffic. Exact reads (GET /api/v1/counters/{key}/exact) query CounterDB for the durably persisted count (~15ms). The exact count lags behind by up to 30 seconds (PersistWorker batch interval).

ShardCache and MasterCache are separate Redis clusters, deliberately isolated. ShardCache handles 10M+ writes/sec (INCR on sub-counters) across a 12-node cluster. MasterCache handles 500K reads/sec (GET on aggregated totals) across a 6-node cluster. Separating them prevents write amplification on the shard path from interfering with read latency.

CounterDB (DynamoDB) provides durable storage with 64 partitions and 3 replicas. It is not on the hot path — PersistWorker writes batch every 30 seconds, and only 5% of reads (exact-count queries) hit it directly. Eventual consistency is acceptable since the primary read path uses MasterCache.