In this article, we will explore the high-level design of a distributed counter system.

This problem is a common choice in system design interviews because it touches on fundamental distributed systems concepts: consistency vs availability trade-offs, hot key management, and the CAP theorem in practice.

Let's start by clarifying the requirements:

Before starting the design, it's important to ask thoughtful questions to uncover hidden assumptions, clarify ambiguities, and define the system's scope more precisely.

Here is an example of how a discussion between the candidate and the interviewer might unfold:

After gathering the details, we can summarize the key system requirements.

To understand the scale of our system, let's make some reasonable assumptions.

Write Volume

Read Volume

Counter Count

Storage

Each counter stores:

Total ≈ 64 bytes per counter

Storage estimate: 50B counters × 64 bytes = 3.2 TB

This fits comfortably in a distributed in-memory store, though we will also need persistent storage for durability.

Hot Counter Challenge

A viral post might receive 10,000 likes per second on a single counter. This is the core challenge: how to handle extreme write concentration on one key.

The distributed counter needs a simple but reliable API for counter operations.

1. Increment Counter

Endpoint: POST /counters/{entity_id}/{counter_type}/increment

Increments the specified counter by a given amount.

Request Parameters

Sample Response

Note: The response does not return the new count to avoid read-after-write latency. Clients should use the read API if they need the current value.

Error Cases

2. Decrement Counter

Endpoint: POST /counters/{entity_id}/{counter_type}/decrement

Decrements the specified counter. Identical structure to increment.

Additional Behavior

Counters may or may not be allowed to go negative, depending on business rules. For like counts, decrement represents an "unlike" and should not go below zero.

3. Read Counter

Endpoint: GET /counters/{entity_id}/{counter_type}

Retrieves the current value of a counter.

Request Parameters

Sample Response

4. Batch Read Counters

Endpoint: POST /counters/batch

Retrieves multiple counters in a single request (for rendering feeds with many posts).

Request Body

Sample Response

At a high level, our distributed counter system must satisfy three core requirements:

The key insight is that writes and reads have very different characteristics. Writes need durability and can tolerate slight delays in visibility. Reads need speed and can tolerate slight staleness. This suggests separating the write and read paths.

Note: Instead of presenting the full architecture at once, we will build it incrementally by addressing one requirement at a time. This approach is easier to follow and mirrors how you would explain the design in an interview.

Let's start with the simplest possible design and identify its limitations.

Naive Approach: Single Database Counter

The most straightforward approach is storing counters in a database and using atomic increment operations.

Why This Fails: All updates to a single counter serialize on one database row, so concurrent updates wait on the row lock. One database also cannot handle millions of QPS, making it a single point of failure.

This approach might work for 100 updates/second, but fails catastrophically for viral content.

Components Needed

Counter Service

A stateless service layer that handles counter operations and routes them appropriately. It receives increment, decrement, and read requests, validates inputs, routes each request to the appropriate storage layer, and handles retries and failures gracefully.

Write Buffer (Redis)

An in-memory buffer that absorbs high-throughput writes before persisting to durable storage. It accepts counter increments with sub-millisecond latency, batches updates for efficient persistence, and provides atomic increment operations.

Persistent Storage (Database)

The durable source of truth for counter values. It stores counter values permanently, survives system restarts and failures, and supports efficient batch updates.

Flow: Incrementing a Counter

This sequence traces a single increment from client request through the Write Buffer to background database sync, showing where each acknowledgment happens along the path.

This approach provides fast writes by decoupling the client-facing operation from durable persistence.

The basic design still has a problem: a single Redis key for a viral post becomes a bottleneck. Redis is single-threaded per key, so thousands of concurrent increments on one key will serialize.

The Hot Key Problem

For example, a celebrity posts a photo and 50,000 users like it within the first second. All 50,000 increments hit the same Redis key. Redis can handle maybe 100,000 ops/sec total, but concentrated on one key, it becomes the bottleneck.

Solution: Sharded Counters

Instead of one counter per entity, split each counter into multiple shards.

How Sharded Counters Work

Write Path:

Read Path:

Dynamic Shard Count

Not all counters need the same number of shards. A new post with 10 likes/day works fine with 1 shard, while a viral post with 10,000 likes/second needs 100 or more shards.

Adaptive Sharding:

Additional Component: Shard Manager

The Shard Manager tracks the shard count per counter, decides when to split or merge shards, and maintains shard metadata.

Summing multiple shards on every read is slow. For a counter with 100 shards, we would need 100 Redis reads, adding significant latency.

Solution: Materialized Read Cache

Maintain a pre-computed total alongside the shards.

Components Needed

Read Cache

A separate cache layer storing pre-aggregated counter totals. It stores the current total for fast O(1) reads, updates asynchronously as writes occur, and serves reads with sub-millisecond latency.

Dual-Path Architecture

Writes and reads travel through completely separate paths, with a background Aggregator acting as the bridge that keeps the Read Cache in sync with the sharded counters.

Write Path

Read Path

Background Aggregation

This introduces a small lag between writes and read visibility, but keeps reads extremely fast.

So far, everything is in Redis. If Redis crashes, we lose counter data. We need durable persistence.

Solution: Write-Ahead Log + Async Persistence

Components Needed

Write-Ahead Log (WAL)

A durable, append-only log that records every counter operation before it is applied. It durably records every increment and decrement, enables recovery after crashes, and supports replay for consistency.

Persistent Database

The long-term storage for counter values. It stores finalized counter values, serves as the source of truth for recovery, and supports efficient batch updates.

Flow: Durable Increment

This sequence shows the ordering that guarantees durability: the WAL is written before Redis is updated, so the operation survives a Redis crash even if the acknowledgment has already been sent to the client.

Recovery Process

Here is the complete architecture combining all requirements:

Core Components Summary

Data Flow Summary

The distributed counter has three distinct storage needs:

Sharded Counter Keys

Each physical shard is stored as a separate Redis string key, with the shard index embedded in the key name so the Counter Service can address any shard directly.

Shard Metadata

Alongside the value shards, a separate Redis hash stores the current shard count and compaction timestamps, giving the Shard Manager the context it needs to decide when to split or merge.

Read Cache Keys

The pre-aggregated total lives under a :total key with a rolling TTL, so reads always see a single integer rather than summing shards on every request.

1. Counters Table

The source of truth for all counter values.

Primary Key: (entity_id, counter_type)

Index: idx_counters_updated on updated_at for finding recently changed counters

2. Counter Events Table (Optional)

For audit trail and analytics, store individual counter events.

Index: idx_events_entity on (entity_id, counter_type, created_at) for time-range queries

Partitioning: Partition by created_at (daily or weekly) for efficient cleanup of old events

Now that we have the high-level architecture and database schema in place, let's dive deeper into some critical design choices.

There are several approaches to implementing distributed counters, each with different trade-offs. Understanding these helps you make informed decisions in an interview.

Architecture 1: Single Counter with Locking

The simplest approach: one row per counter with atomic updates.

How It Works

A single SQL UPDATE with an inline arithmetic expression is all it takes: the database applies the increment atomically by acquiring a row-level lock for the duration of the write.

Best For

Low-throughput counters (< 100 updates/second), systems requiring strong consistency.

Architecture 2: Sharded Counters

Split each logical counter into multiple physical shards.

How It Works

Write:

Read:

Best For

High-throughput counters, viral content scenarios, when eventual consistency is acceptable.

Architecture 3: CRDT Counters

Conflict-free Replicated Data Types (CRDTs) are designed for distributed systems where replicas may diverge and later merge.

How It Works

A G-Counter (Grow-only Counter) uses per-node counters:

Each node only increments its own slot. Merging takes the maximum of each slot:

A PN-Counter (Positive-Negative Counter) supports decrements by maintaining two G-Counters:

Best For

Multi-datacenter deployments, systems prioritizing availability over consistency, offline-capable applications.

Architecture 4: Event Sourcing

Instead of storing counter values, store the events that modify them.

How It Works

Rather than maintaining a running total, the system appends immutable increment and decrement records. The current counter value is derived by replaying the log, or read from a materialized view that is kept in sync with new events.

Best For

Systems requiring audit trails, undo functionality, or time-travel queries.

Summary and Recommendation

Recommendation: Use Sharded Counters as the default for high-scale systems. The complexity is manageable, and it handles hot keys well. Add event sourcing if you need audit trails for compliance or analytics.

One of the most important decisions in distributed counter design is the consistency guarantee you provide.

Strong Consistency

Every read returns the most recent write. If user A increments a counter and user B reads it immediately after, B sees the incremented value.

Implementation

Use a single authoritative database with synchronous replication, route all reads through the primary database, and apply updates using distributed transactions or consensus protocols.

Trade-offs

When to Use

Eventual Consistency

Reads may return stale values, but all replicas will eventually converge to the same value if updates stop.

Implementation

Write to a local replica and propagate asynchronously. Reads are served from any replica, possibly stale, and background processes reconcile differences.

Trade-offs

When to Use

Read-Your-Writes Consistency

A middle ground: users always see their own updates, but may see stale data from others.

Implementation

To achieve read-your-writes, the service fetches the cached total and then adds any pending delta for the requesting user before returning the value, without touching the shared shard layer.

Trade-offs

When to Use

Consistency in Our Design

Our architecture supports configurable consistency:

When a post goes viral, it might receive 100,000 likes per second. Even with sharding, this is challenging.

The Thundering Herd Problem

A viral event does not produce a steady load; it produces an exponential spike in the first few seconds, overwhelming whatever capacity was provisioned for normal traffic.

Solution 1: Aggressive Sharding

Automatically increase shard count when write rate exceeds thresholds.

Scaling Example

Solution 2: Write Batching at Client

Instead of sending every like individually, batch them at the edge.

Implementation:

Solution 3: Probabilistic Counting

For extremely high-throughput counters where exact accuracy is not needed, use probabilistic counting.

Approach: Only record a sample of events and multiply:

Trade-off: Less accurate for small counts, but reduces write load by 100x.

YouTube famously uses this for view counts on viral videos.

Solution 4: Pre-Provisioned Hot Keys

For predictable viral events (Super Bowl, product launches), pre-provision extra capacity.

A critical challenge in distributed systems: what happens when a request is retried?

The Problem

Network failures during an in-flight request create ambiguity: the client cannot tell whether the server applied the update before the connection dropped, so retrying risks applying the same increment twice.

Solution 1: Idempotency Keys

Clients include a unique key with each request. The server rejects duplicates.

Solution 2: Actor-Based Deduplication

For user-initiated counters (likes), track which users have already acted.

This naturally prevents duplicate likes from the same user.

Solution 3: Event Log Deduplication

If using event sourcing, deduplicate at the event ingestion layer.

Trade-offs

Since reads are typically 10x more frequent than writes, optimizing the read path is critical.

Strategy 1: Aggressive Caching with Short TTL

Cache counter values with a short TTL, accepting slight staleness.

Trade-off: Values may be up to 5 seconds stale.

Strategy 2: Background Refresh

Proactively refresh popular counters before cache expires.

This ensures hot counters are always fresh in cache.

Strategy 3: Approximate Counts for Display

For UI display, exact counts often do not matter. Show human-friendly approximations.

When displaying "3.2M likes", being off by a few thousand does not matter.

Strategy 4: Batch Reads for Feeds

When rendering a feed with many posts, batch all counter reads into one request.

This fetches 100 counters in 1 Redis round-trip instead of 100.

What happens when parts of the system fail?

Redis Failure

Scenario: Redis cluster becomes unavailable

Impact: Cannot write or read counter values

Mitigation

Write-Ahead Log Failure

Scenario: Kafka becomes unavailable

Impact: Cannot durably record writes

Mitigation

Database Failure

Scenario: Primary database is down

Impact: Cannot persist counter values, cannot recover from cold start

Mitigation

Recovery Process

After a failure, the system must reconcile state: