{"title":"Design Distributed Counter","description":null,"content":"\n> **QUESTION**\n>\n> #### What is a Distributed Counter?\n>\n> A distributed counter is a system that maintains numeric counts across multiple servers while handling high-throughput increments and decrements.\n>\n> The core idea seems simple: track how many times something happened. But at scale, counting becomes surprisingly complex. When millions of users simultaneously like a viral post or view a trending video, the counter must handle massive concurrent updates without becoming a bottleneck or losing accuracy.\n>\n> **Popular Examples:** YouTube view counts, Twitter/X like counts, Reddit upvotes, Instagram follower counts\n\n\nIn this article, we will explore the **high-level design of a distributed counter system**.\n\nThis 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.\n\nLet's start by clarifying the requirements:\n\n---\n\n# 1. Clarifying Requirements\n\nBefore starting the design, it's important to ask thoughtful questions to uncover hidden assumptions, clarify ambiguities, and define the system's scope more precisely.\n\nHere is an example of how a discussion between the candidate and the interviewer might unfold:\n\n\n> **DISCUSSION**\n>\n> **Candidate:** \"What is the expected scale? How many counter updates per second should the system handle?\"\n>\n> **Interviewer:** \"Let's design for a social media platform scale. Assume 1 million counter updates per second during peak, with some counters being extremely hot, like a viral post getting thousands of likes per second.\"\n>\n> **Candidate:** \"What types of operations do we need to support? Just increments, or also decrements and reads?\"\n>\n> **Interviewer:** \"We need increment, decrement, and read operations. Think of use cases like likes (increment/decrement) and view counts (increment only).\"\n>\n> **Candidate:** \"What consistency guarantees do we need? Should reads always return the exact count, or is eventual consistency acceptable?\"\n>\n> **Interviewer:** \"Eventual consistency is acceptable for most use cases. Users can tolerate seeing '1.2M likes' even if the actual count is 1,200,042. However, we should not lose any updates.\"\n>\n> **Candidate:** \"Should the system support multiple counters? For example, each post has its own like count, view count, and share count?\"\n>\n> **Interviewer:** \"Yes, we need to support billions of independent counters. Each entity (post, video, user profile) can have multiple counter types.\"\n>\n> **Candidate:** \"What are the latency requirements for reads and writes?\"\n>\n> **Interviewer:** \"Writes should be acknowledged quickly, under 50ms. Reads should be very fast, under 10ms, since they happen on every page load.\"\n>\n> **Candidate:** \"Do we need to support any analytics, like 'likes in the last hour' or historical counts?\"\n>\n> **Interviewer:** \"For this interview, let's focus on current counts only. Time-windowed analytics can be discussed as an extension.\"\n\n\nAfter gathering the details, we can summarize the key system requirements.\n\n## 1.1 Functional Requirements\n\n- **Increment:** Increase a counter's value by a specified amount (default: 1).\n- **Decrement:** Decrease a counter's value by a specified amount (default: 1).\n- **Read:** Retrieve the current value of a counter.\n- **Multi-Counter Support:** Each entity can have multiple counter types (likes, views, shares).\n\n## 1.2 Non-Functional Requirements\n\n- **High Availability:** The system must remain operational even during partial failures. Target 99.99% availability.\n- **High Throughput:** Handle millions of updates per second across all counters.\n- **Hot Key Resilience:** Individual counters must handle thousands of concurrent updates (viral content scenario).\n- **Low Latency:** Writes acknowledged in < 50ms, reads served in < 10ms.\n- **Durability:** No updates should be lost, even during failures.\n- **Eventual Consistency:** Reads may return slightly stale values, but will eventually converge to the correct count.\n\n---\n\n# 2. Back-of-the-Envelope Estimation\n\nTo understand the scale of our system, let's make some reasonable assumptions.\n\n#### **Write Volume**\n\n- Counter updates: **1 million per second** (peak)\n- Average updates: **300,000 per second** (steady state)\n- Peak write load (3x factor): **3 million per second** (viral event)\n\n#### **Read Volume**\n\n- Assume a **10:1 read/write ratio** (counters displayed on every page view)\n- Average read QPS: **3 million per second**\n- Peak read QPS: **10 million per second**\n\n#### **Counter Count**\n\n- Total entities (posts, videos, users): **10 billion**\n- Counter types per entity: **5** (likes, views, shares, comments, saves)\n- Total counters: **50 billion**\n\n#### **Storage**\n\nEach counter stores:\n\n- Entity ID (~16 bytes)\n- Counter type (~8 bytes)\n- Counter value (~8 bytes)\n- Metadata (~32 bytes)\n\nTotal ≈ **64 bytes per counter**\n\n**Storage estimate:** `50B counters × 64 bytes = 3.2 TB`\n\nThis fits comfortably in a distributed in-memory store, though we will also need persistent storage for durability.\n\n#### **Hot Counter Challenge**\n\nA viral post might receive:\n\n- **10,000 likes per second** on a single counter\n- This is the core challenge: how to handle extreme write concentration on one key\n\n---\n\n# 3. Core APIs\n\nThe distributed counter needs a simple but robust API for counter operations.\n\n### **1. Increment Counter**\n\n#### Endpoint: `POST /counters/{entity_id}/{counter_type}/increment`\n\nIncrements the specified counter by a given amount.\n\n##### **Request Parameters:**\n\n- **entity_id** _(required)_: The unique identifier for the entity (post ID, video ID, user ID).\n- **counter_type** _(required)_: The type of counter (likes, views, shares).\n- **amount** _(optional)_: The increment amount. Default is 1.\n- **idempotency_key** _(optional)_: Unique key to prevent duplicate increments from retries.\n\n##### **Sample Response:**\n\n- **success**: Boolean indicating the operation was accepted.\n- **counter_id**: The full counter identifier.\n- **acknowledged**: Whether the increment was durably recorded.\n\nNote: 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.\n\n##### **Error Cases:**\n\n- `400 Bad Request`: Invalid entity_id or counter_type.\n- `429 Too Many Requests`: Client is being rate limited.\n- `503 Service Unavailable`: System is overloaded.\n\n### **2. Decrement Counter**\n\n#### Endpoint: `POST /counters/{entity_id}/{counter_type}/decrement`\n\nDecrements the specified counter. Identical structure to increment.\n\n##### **Additional Behavior:**\n\n- Counters may or may not be allowed to go negative, depending on business rules.\n- For like counts, decrement represents \"unlike\" and should not go below zero.\n\n### **3. Read Counter**\n\n#### Endpoint: `GET /counters/{entity_id}/{counter_type}`\n\nRetrieves the current value of a counter.\n\n##### **Request Parameters:**\n\n- **entity_id** _(required)_: The entity identifier.\n- **counter_type** _(required)_: The counter type.\n- **consistency** _(optional)_: \"eventual\" (default, fast) or \"strong\" (slower, accurate).\n\n##### **Sample Response:**\n\n- **value**: The current counter value.\n- **last_updated**: When the counter was last modified.\n- **is_approximate**: Whether the value might be slightly stale.\n\n### **4. Batch Read Counters**\n\n#### Endpoint: `POST /counters/batch`\n\nRetrieves multiple counters in a single request (for rendering feeds with many posts).\n\n##### **Request Body:**\n\n\n```json\n{\n \"counters\": [\n {\"entity_id\": \"post_123\", \"counter_type\": \"likes\"},\n {\"entity_id\": \"post_123\", \"counter_type\": \"views\"},\n {\"entity_id\": \"post_456\", \"counter_type\": \"likes\"}\n ]\n}\n```\n\n\n##### **Sample Response:**\n\n\n```json\n{\n \"counters\": [\n {\"entity_id\": \"post_123\", \"counter_type\": \"likes\", \"value\": 15420},\n {\"entity_id\": \"post_123\", \"counter_type\": \"views\", \"value\": 89234},\n {\"entity_id\": \"post_456\", \"counter_type\": \"likes\", \"value\": 342}\n ]\n}\n```\n\n\n---\n\n# 4. High-Level Design\n\nAt a high level, our distributed counter system must satisfy three core requirements:\n\n1. **High-Throughput Writes:** Accept millions of counter updates per second.\n2. **Low-Latency Reads:** Serve counter values quickly for page rendering.\n3. **Hot Key Handling:** Prevent any single counter from becoming a bottleneck.\n\nThe 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.\n\n**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.\n\n## 4.1 Requirement 1: Basic Counter Operations\n\nLet's start with the simplest possible design and identify its limitations.\n\n### Naive Approach: Single Database Counter\n\nThe most straightforward approach is storing counters in a database and using atomic increment operations.\n\n\n```sql\nUPDATE counters SET value = value + 1 WHERE entity_id = ? AND counter_type = ?\n```\n\n\n**Why This Fails:**\n\n- **Write Bottleneck:** All updates to a single counter serialize on one database row\n- **Lock Contention:** Concurrent updates cause lock waits\n- **Single Point of Failure:** One database cannot handle millions of QPS\n\nThis approach might work for 100 updates/second, but fails catastrophically for viral content.\n\n### Components Needed\n\n#### **Counter Service**\n\nA stateless service layer that handles counter operations and routes them appropriately.\n\n**Responsibilities:**\n\n- Receive increment/decrement/read requests\n- Validate inputs\n- Route requests to the appropriate storage layer\n- Handle retries and failures gracefully\n\n#### **Write Buffer (Redis)**\n\nAn in-memory buffer that absorbs high-throughput writes before persisting to durable storage.\n\n**Responsibilities:**\n\n- Accept counter increments with sub-millisecond latency\n- Batch updates for efficient persistence\n- Provide atomic increment operations\n\n#### **Persistent Storage (Database)**\n\nThe durable source of truth for counter values.\n\n**Responsibilities:**\n\n- Store counter values permanently\n- Survive system restarts and failures\n- Support efficient batch updates\n\n### Flow: Incrementing a Counter\n\n\n```mermaid\nsequenceDiagram\n participant Client\n participant CS as Counter Service\n participant Redis as Write Buffer\n participant DB as Database\n\n Client->>CS: POST /increment (entity_id, counter_type)\n CS->>Redis: INCRBY counter_key 1\n Redis-->>CS: OK\n CS-->>Client: {success: true}\n\n Note over Redis,DB: Background sync (every few seconds)\n Redis->>DB: Batch UPDATE counters\n DB-->>Redis: OK\n Redis->>Redis: Reset buffered deltas\n```\n\n\n1. Client sends an increment request to the **Counter Service**.\n2. The service performs an atomic `INCRBY` on the **Write Buffer** (Redis).\n3. The service immediately acknowledges success to the client.\n4. A background process periodically flushes accumulated deltas to the **Database**.\n\nThis approach provides fast writes by decoupling the client-facing operation from durable persistence.\n\n---\n\n## 4.2 Requirement 2: Handling Hot Keys\n\nThe 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.\n\n### The Hot Key Problem\n\n**Example:**\n\n- A celebrity posts a photo\n- 50,000 users like it within the first second\n- All 50,000 increments hit the same Redis key\n- Redis can handle maybe 100,000 ops/sec total, but concentrated on one key, it becomes the bottleneck\n\n### Solution: Sharded Counters\n\nInstead of one counter per entity, split each counter into multiple shards.\n\n\n```shell\nOriginal: post_123:likes = 50,000\n\nSharded:\n post_123:likes:shard_0 = 12,500\n post_123:likes:shard_1 = 12,500\n post_123:likes:shard_2 = 12,500\n post_123:likes:shard_3 = 12,500\n\nTotal = sum of all shards = 50,000\n```\n\n\n#### How Sharded Counters Work\n\n**Write Path:**\n\n1. Hash the request (using client ID, request ID, or random selection)\n2. Route to one of N shards: `shard = hash(request_id) % num_shards`\n3. Increment only that shard\n\n**Read Path:**\n\n1. Read all shards for the counter\n2. Sum the values\n3. Return the total\n\n\n```mermaid\nflowchart TB\n subgraph Writes[Write Requests]\n W1[Increment 1]\n W2[Increment 2]\n W3[Increment 3]\n W4[Increment 4]\n end\n\n subgraph Shards[Counter Shards]\n S0[Shard 0
value: 125]\n S1[Shard 1
value: 118]\n S2[Shard 2
value: 132]\n S3[Shard 3
value: 125]\n end\n\n subgraph Read[Read Request]\n R[Sum: 500]\n end\n\n W1 -->|hash % 4 = 0| S0\n W2 -->|hash % 4 = 1| S1\n W3 -->|hash % 4 = 2| S2\n W4 -->|hash % 4 = 3| S3\n\n S0 --> R\n S1 --> R\n S2 --> R\n S3 --> R\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef secondary fill:#38d9a9,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n\n class W1,W2,W3,W4 primary\n class S0,S1,S2,S3 orange\n class R secondary\n```\n\n\n#### Dynamic Shard Count\n\nNot all counters need the same number of shards:\n\n- A new post with 10 likes/day: 1 shard is sufficient\n- A viral post with 10,000 likes/second: needs 100+ shards\n\n**Adaptive Sharding:**\n\n1. Start all counters with 1 shard\n2. Monitor write rate per counter\n3. If write rate exceeds threshold, add more shards\n4. Reduce shards when activity decreases (during background compaction)\n\n### Additional Component: Shard Manager\n\n**Responsibilities:**\n\n- Track shard count per counter\n- Decide when to split or merge shards\n- Maintain shard metadata\n\n---\n\n## 4.3 Requirement 3: Fast Reads\n\nSumming multiple shards on every read is slow. For a counter with 100 shards, we would need 100 Redis reads, adding significant latency.\n\n### Solution: Materialized Read Cache\n\nMaintain a pre-computed total alongside the shards.\n\n#### Components Needed\n\n#### **Read Cache**\n\nA separate cache layer storing pre-aggregated counter totals.\n\n**Responsibilities:**\n\n- Store current total for fast O(1) reads\n- Update asynchronously as writes occur\n- Serve reads with sub-millisecond latency\n\n### Dual-Path Architecture\n\n\n```mermaid\nflowchart LR\n subgraph WritePath[Write Path]\n W[Write Request]\n WS[Counter Service]\n Shard[Sharded Counter
in Redis]\n end\n\n subgraph ReadPath[Read Path]\n R[Read Request]\n RS[Counter Service]\n Cache[Read Cache]\n end\n\n subgraph Sync[Background Sync]\n Agg[Aggregator]\n end\n\n W --> WS\n WS --> Shard\n\n R --> RS\n RS --> Cache\n\n Shard -.->|Periodic aggregation| Agg\n Agg -.->|Update total| Cache\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef secondary fill:#38d9a9,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n\n class W,R primary\n class WS,RS secondary\n class Shard,Cache purple\n class Agg orange\n```\n\n\n**Write Path:**\n\n1. Increment goes to a random shard (distributed load)\n2. Write is acknowledged immediately\n\n**Read Path:**\n\n1. Read goes directly to the Read Cache\n2. Returns pre-aggregated total in O(1)\n\n**Background Aggregation:**\n\n1. Periodically (every 1-5 seconds), sum all shards\n2. Update the Read Cache with the new total\n\nThis introduces a small lag between writes and read visibility, but keeps reads extremely fast.\n\n---\n\n## 4.4 Requirement 4: Durability\n\nSo far, everything is in Redis. If Redis crashes, we lose counter data. We need durable persistence.\n\n### Solution: Write-Ahead Log + Async Persistence\n\n#### Components Needed\n\n#### **Write-Ahead Log (WAL)**\n\nA durable, append-only log that records every counter operation before it is applied.\n\n**Responsibilities:**\n\n- Durably record every increment/decrement\n- Enable recovery after crashes\n- Support replay for consistency\n\n#### **Persistent Database**\n\nThe long-term storage for counter values.\n\n**Responsibilities:**\n\n- Store finalized counter values\n- Serve as source of truth for recovery\n- Support efficient batch updates\n\n### Flow: Durable Increment\n\n\n```mermaid\nsequenceDiagram\n participant Client\n participant CS as Counter Service\n participant WAL as Write-Ahead Log\n participant Redis as Write Buffer\n participant DB as Database\n\n Client->>CS: Increment request\n CS->>WAL: Append operation\n WAL-->>CS: Logged (durable)\n CS->>Redis: INCRBY (fast)\n Redis-->>CS: OK\n CS-->>Client: Success\n\n Note over WAL,DB: Background persistence\n WAL->>DB: Batch apply operations\n DB-->>WAL: Applied\n WAL->>WAL: Truncate applied entries\n```\n\n\n1. Write operation is first appended to the **WAL** (durable, fast append-only write)\n2. Then applied to **Redis** (fast, in-memory)\n3. Client receives success acknowledgment\n4. Background process applies WAL entries to the **Database** in batches\n5. Applied WAL entries are truncated\n\n**Recovery Process:**\n\n1. On startup, load counter values from Database\n2. Replay any WAL entries not yet applied to Database\n3. Resume normal operation\n\n---\n\n## 4.5 Putting It All Together\n\nHere is the complete architecture combining all requirements:\n\n\n```mermaid\nflowchart TB\n subgraph Clients\n C[Clients]\n end\n\n subgraph Edge[Load Balancing]\n LB[Load Balancer]\n end\n\n subgraph Services[Counter Service Layer]\n CS1[Counter Service 1]\n CS2[Counter Service 2]\n CS3[Counter Service 3]\n end\n\n subgraph WriteLayer[Write Layer]\n WAL[(Write-Ahead Log
Kafka)]\n Redis1[(Redis Cluster
Sharded Counters)]\n end\n\n subgraph ReadLayer[Read Layer]\n ReadCache[(Read Cache
Pre-aggregated)]\n end\n\n subgraph Persistence[Persistence Layer]\n Agg[Aggregator
Service]\n DB[(Database
Source of Truth)]\n end\n\n C --> LB\n LB --> CS1\n LB --> CS2\n LB --> CS3\n\n CS1 -->|Writes| WAL\n CS2 -->|Writes| WAL\n CS3 -->|Writes| WAL\n\n WAL --> Redis1\n\n CS1 -->|Reads| ReadCache\n CS2 -->|Reads| ReadCache\n CS3 -->|Reads| ReadCache\n\n Redis1 -.->|Aggregate| Agg\n Agg -.->|Update| ReadCache\n Agg -.->|Persist| DB\n\n classDef primary fill:#00ceff,stroke:#000,color:#000\n classDef secondary fill:#38d9a9,stroke:#000,color:#000\n classDef purple fill:#9775fa,stroke:#000,color:#000\n classDef orange fill:#ffa94d,stroke:#000,color:#000\n classDef green fill:#69db7c,stroke:#000,color:#000\n\n class C,CS1,CS2,CS3 primary\n class LB secondary\n class WAL,Redis1,ReadCache purple\n class Agg orange\n class DB green\n```\n\n\n### Core Components Summary\n\n\n| ##### Component | ##### Purpose |\n| --- | --- |\n| **Load Balancer** | Distributes requests across Counter Service instances |\n| **Counter Service** | Stateless service handling increment/decrement/read operations |\n| **Write-Ahead Log** | Durable log for all write operations (Kafka or similar) |\n| **Redis Cluster** | In-memory storage for sharded counters |\n| **Read Cache** | Pre-aggregated counter totals for fast reads |\n| **Aggregator Service** | Background service that sums shards and updates read cache |\n| **Database** | Persistent storage for counter values |\n\n\n### Data Flow Summary\n\n\n| ##### Operation | ##### Path | ##### Latency |\n| --- | --- | --- |\n| **Write** | Client → Service → WAL → Redis | < 50ms |\n| **Read** | Client → Service → Read Cache | < 10ms |\n| **Aggregation** | Redis → Aggregator → Read Cache | Every 1-5 sec |\n| **Persistence** | WAL → Database | Every 10-60 sec |\n\n\n---\n\n# 5. Database Design\n\n## 5.1 Storage Requirements\n\nThe distributed counter has three distinct storage needs:\n\n1. **Hot Storage (Redis):** Sharded counters for high-throughput writes\n2. **Read Cache (Redis/Memcached):** Pre-aggregated totals for fast reads\n3. **Cold Storage (Database):** Durable, persistent counter values\n\n## 5.2 Redis Schema\n\n### Sharded Counter Keys\n\n\n```shell\nKey Pattern: counter:{entity_id}:{counter_type}:shard:{shard_id}\nType: String (integer value)\nTTL: None (persistent)\n\nExample:\n counter:post_123:likes:shard:0 = \"12500\"\n counter:post_123:likes:shard:1 = \"12498\"\n counter:post_123:likes:shard:2 = \"12502\"\n counter:post_123:likes:shard:3 = \"12500\"\n```\n\n\n### Shard Metadata\n\n\n```shell\nKey Pattern: counter:{entity_id}:{counter_type}:meta\nType: Hash\nFields:\n - num_shards: Current number of shards\n - created_at: When the counter was created\n - last_compacted: When shards were last merged\n\nExample:\n counter:post_123:likes:meta = {\n \"num_shards\": \"4\",\n \"created_at\": \"1703980800\",\n \"last_compacted\": \"1703990000\"\n }\n```\n\n\n### Read Cache Keys\n\n\n```shell\nKey Pattern: counter:{entity_id}:{counter_type}:total\nType: String (integer value)\nTTL: 300 seconds (refreshed by aggregator)\n\nExample:\n counter:post_123:likes:total = \"50000\"\n```\n\n\n## 5.3 Database Schema (PostgreSQL)\n\n### **1. Counters Table**\n\nThe source of truth for all counter values.\n\n\n| ##### Field | ##### Type | ##### Description |\n| --- | --- | --- |\n| `entity_id` | VARCHAR(64) | Entity identifier (PK part 1) |\n| `counter_type` | VARCHAR(32) | Counter type: likes, views, shares (PK part 2) |\n| `value` | BIGINT | Current counter value |\n| `version` | BIGINT | Optimistic locking version |\n| `updated_at` | TIMESTAMP | Last update timestamp |\n| `created_at` | TIMESTAMP | Creation timestamp |\n\n\n**Primary Key:** (`entity_id`, `counter_type`)\n\n**Index:** `idx_counters_updated` on `updated_at` for finding recently changed counters\n\n### **2. Counter Events Table (Optional)**\n\nFor audit trail and analytics, store individual counter events.\n\n\n| ##### Field | ##### Type | ##### Description |\n| --- | --- | --- |\n| `id` | UUID | Unique event identifier (PK) |\n| `entity_id` | VARCHAR(64) | Entity identifier |\n| `counter_type` | VARCHAR(32) | Counter type |\n| `delta` | INTEGER | Change amount (+1, -1, etc.) |\n| `actor_id` | VARCHAR(64) | Who triggered the change (user ID) |\n| `idempotency_key` | VARCHAR(64) | For deduplication |\n| `created_at` | TIMESTAMP | When the event occurred |\n\n\n**Index:** `idx_events_entity` on (`entity_id`, `counter_type`, `created_at`) for time-range queries\n\n**Partitioning:** Partition by `created_at` (daily or weekly) for efficient cleanup of old events\n\n---\n\n# 6. Design Deep Dive\n\nNow that we have the high-level architecture and database schema in place, let's dive deeper into some critical design choices.\n\n## 6.1 Counter Architectures\n\nThere are several approaches to implementing distributed counters, each with different trade-offs. Understanding these helps you make informed decisions in an interview.\n\n### **Architecture 1: Single Counter with Locking**\n\nThe simplest approach: one row per counter with atomic updates.\n\n#### How It Works\n\n\n```sql\n-- Atomic increment with row lock\nUPDATE counters\nSET value = value + 1, updated_at = NOW()\nWHERE entity_id = 'post_123' AND counter_type = 'likes';\n```\n\n\n#### Pros\n\n- **Simple:** Easy to implement and understand\n- **Strongly consistent:** Reads always return accurate values\n- **Transactional:** Easy to combine with other operations\n\n#### Cons\n\n- **Lock contention:** Concurrent updates serialize on the row lock\n- **Limited throughput:** Maybe 1,000-5,000 updates/second per counter\n- **Database bottleneck:** Database becomes single point of failure\n\n#### Best For\n\nLow-throughput counters (< 100 updates/second), systems requiring strong consistency.\n\n### **Architecture 2: Sharded Counters**\n\nSplit each logical counter into multiple physical shards.\n\n#### How It Works\n\n**Write:**\n\n\n```python\nshard_id = hash(request_id) % num_shards\nredis.incr(f\"counter:{entity_id}:{type}:shard:{shard_id}\")\n```\n\n\n**Read:**\n\n\n```python\ntotal = 0\nfor shard_id in range(num_shards):\n total += redis.get(f\"counter:{entity_id}:{type}:shard:{shard_id}\")\nreturn total\n```\n\n\n#### Pros\n\n- **Horizontal scalability:** Each shard handles a fraction of the load\n- **No lock contention:** Different shards can be updated in parallel\n- **Adjustable:** Add shards for hot counters, reduce for cold ones\n\n#### Cons\n\n- **Read amplification:** Must read all shards to get total\n- **Eventual consistency:** Aggregated totals may be slightly stale\n- **Complexity:** Managing shard counts adds operational overhead\n\n#### Best For\n\nHigh-throughput counters, viral content scenarios, when eventual consistency is acceptable.\n\n### **Architecture 3: CRDT Counters**\n\nConflict-free Replicated Data Types (CRDTs) are designed for distributed systems where replicas may diverge and later merge.\n\n#### How It Works\n\nA **G-Counter** (Grow-only Counter) uses per-node counters:\n\n\n```shell\nNode A: {A: 5, B: 0, C: 0}\nNode B: {A: 0, B: 3, C: 0}\nNode C: {A: 0, B: 0, C: 7}\n\nTotal = 5 + 3 + 7 = 15\n```\n\n\nEach node only increments its own slot. Merging takes the maximum of each slot:\n\n\n```shell\nMerge(NodeA, NodeB) = {A: 5, B: 3, C: 0}\n```\n\n\nA **PN-Counter** (Positive-Negative Counter) supports decrements by maintaining two G-Counters:\n\n- P-Counter for increments\n- N-Counter for decrements\n- Value = sum(P) - sum(N)\n\n#### Pros\n\n- **Partition tolerant:** Works even when nodes cannot communicate\n- **No coordination:** No locks, no consensus needed\n- **Mathematically guaranteed convergence:** All replicas eventually agree\n\n#### Cons\n\n- **Memory overhead:** Stores state per node, not just a single value\n- **Complexity:** More complex to implement correctly\n- **No strong consistency:** Cannot provide \"exactly X\" guarantees\n\n#### Best For\n\nMulti-datacenter deployments, systems prioritizing availability over consistency, offline-capable applications.\n\n### **Architecture 4: Event Sourcing**\n\nInstead of storing counter values, store the events that modify them.\n\n#### How It Works\n\n\n```shell\nEvents for post_123:likes:\n {event: \"increment\", actor: \"user_1\", time: \"10:00:01\"}\n {event: \"increment\", actor: \"user_2\", time: \"10:00:02\"}\n {event: \"decrement\", actor: \"user_1\", time: \"10:00:05\"}\n {event: \"increment\", actor: \"user_3\", time: \"10:00:06\"}\n\nCurrent value = 1 + 1 - 1 + 1 = 2\n```\n\n\n#### Pros\n\n- **Complete audit trail:** Know exactly who did what and when\n- **Flexible queries:** Can compute count at any point in time\n- **Enables undo:** Can reverse specific actions\n\n#### Cons\n\n- **Read complexity:** Must replay events or maintain materialized views\n- **Storage growth:** Events accumulate forever (or require snapshotting)\n- **Performance:** Counting millions of events is slow without optimization\n\n#### Best For\n\nSystems requiring audit trails, undo functionality, or time-travel queries.\n\n### Summary and Recommendation\n\n\n| ##### Architecture | ##### Throughput | ##### Consistency | ##### Complexity | ##### Best For |\n| --- | --- | --- | --- | --- |\n| **Single Counter** | Low | Strong | Simple | Low-traffic counters |\n| **Sharded Counters** | High | Eventual | Medium | Viral content, social media |\n| **CRDT Counters** | High | Eventual | High | Multi-datacenter, offline |\n| **Event Sourcing** | Medium | Eventual | High | Audit trails, analytics |\n\n\n**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.\n\n---\n\n## 6.2 Consistency Models\n\nOne of the most important decisions in distributed counter design is the consistency guarantee you provide.\n\n### Strong Consistency\n\nEvery read returns the most recent write. If user A increments a counter and user B reads it immediately after, B sees the incremented value.\n\n#### Implementation\n\n- Use a single authoritative database with synchronous replication\n- All reads go through the primary database\n- Updates use distributed transactions or consensus protocols\n\n#### Trade-offs\n\n- **Latency:** Reads are slower (database round-trip)\n- **Availability:** System unavailable if primary is down\n- **Throughput:** Limited by single-node write capacity\n\n#### When to Use\n\n- Financial counters (account balances, inventory stock)\n- Counters used for access control decisions\n- Regulatory requirements demand accuracy\n\n### Eventual Consistency\n\nReads may return stale values, but all replicas will eventually converge to the same value if updates stop.\n\n#### Implementation\n\n- Write to local replica, propagate asynchronously\n- Reads served from any replica (possibly stale)\n- Background processes reconcile differences\n\n#### Trade-offs\n\n- **Latency:** Very fast reads and writes\n- **Availability:** System remains available during partitions\n- **Accuracy:** Values may be temporarily incorrect\n\n#### When to Use\n\n- Social media metrics (likes, views, shares)\n- Analytics dashboards\n- Any counter where \"approximately correct\" is acceptable\n\n### Read-Your-Writes Consistency\n\nA middle ground: users always see their own updates, but may see stale data from others.\n\n#### Implementation\n\n\n```python\ndef read_counter(entity_id, counter_type, user_id):\n # Get cached value\n cached_value = read_cache.get(entity_id, counter_type)\n\n # Check if user has pending writes\n pending_delta = get_user_pending_writes(user_id, entity_id, counter_type)\n\n return cached_value + pending_delta\n```\n\n\n#### Trade-offs\n\n- **User experience:** Users see their actions reflected immediately\n- **Complexity:** Must track per-user pending writes\n- **Partial consistency:** Other users' updates may be delayed\n\n#### When to Use\n\n- Like buttons (user should see their like immediately)\n- Any counter where user feedback is important\n\n### Consistency in Our Design\n\nOur architecture supports **configurable consistency**:\n\n\n| ##### Read Type | ##### How It Works | ##### Latency | ##### Use Case |\n| --- | --- | --- | --- |\n| **Eventual** | Read from Read Cache | < 5ms | Page rendering, feeds |\n| **Read-Your-Writes** | Cache + user's pending delta | < 10ms | After user action |\n| **Strong** | Sum all shards directly | < 50ms | Admin dashboards, debugging |\n\n\n---\n\n## 6.3 Handling Viral Content\n\nWhen a post goes viral, it might receive 100,000 likes per second. Even with sharding, this is challenging.\n\n### The Thundering Herd Problem\n\n\n```shell\nCelebrity posts photo at 10:00:00\n 10:00:01 - 1,000 likes\n 10:00:02 - 5,000 likes\n 10:00:03 - 20,000 likes\n 10:00:04 - 50,000 likes\n 10:00:05 - 100,000 likes ← System overload\n```\n\n\n### Solution 1: Aggressive Sharding\n\nAutomatically increase shard count when write rate exceeds thresholds.\n\n\n```python\ndef maybe_increase_shards(entity_id, counter_type):\n current_rate = get_write_rate(entity_id, counter_type) # writes/sec\n current_shards = get_shard_count(entity_id, counter_type)\n\n # Target: max 1000 writes per shard per second\n target_shards = ceil(current_rate / 1000)\n\n if target_shards > current_shards:\n add_shards(entity_id, counter_type, target_shards - current_shards)\n```\n\n\n**Scaling Example:**\n\n- 1,000 writes/sec → 1 shard\n- 10,000 writes/sec → 10 shards\n- 100,000 writes/sec → 100 shards\n\n### Solution 2: Write Batching at Client\n\nInstead of sending every like individually, batch them at the edge.\n\n\n```shell\nWithout batching:\n 100,000 individual requests → 100,000 Redis operations\n\nWith batching (100ms window):\n 100,000 likes → grouped into ~1,000 batches\n Each batch: INCRBY counter 100 (average)\n Result: 1,000 Redis operations\n```\n\n\n**Implementation:**\n\n\n```python\n# At edge/gateway level\nclass CounterBatcher:\n def __init__(self, flush_interval_ms=100):\n self.pending = defaultdict(int) # counter_key -> delta\n self.flush_interval = flush_interval_ms\n\n def increment(self, entity_id, counter_type, amount=1):\n key = f\"{entity_id}:{counter_type}\"\n self.pending[key] += amount\n\n def flush(self):\n for key, delta in self.pending.items():\n redis.incrby(f\"counter:{key}:shard:{random_shard()}\", delta)\n self.pending.clear()\n```\n\n\n### Solution 3: Probabilistic Counting\n\nFor extremely high-throughput counters where exact accuracy is not needed, use probabilistic counting.\n\n**Approach:** Only record a sample of events and multiply:\n\n\n```python\nSAMPLE_RATE = 0.01 # 1% sampling\n\ndef increment_sampled(entity_id, counter_type):\n if random.random() < SAMPLE_RATE:\n redis.incrby(f\"counter:{entity_id}:{counter_type}\", int(1 / SAMPLE_RATE))\n```\n\n\n**Trade-off:** Less accurate for small counts, but reduces write load by 100x.\n\nYouTube famously uses this for view counts on viral videos.\n\n### Solution 4: Pre-Provisioned Hot Keys\n\nFor predictable viral events (Super Bowl, product launches), pre-provision extra capacity.\n\n\n```python\ndef prepare_for_viral_event(entity_id, counter_type, expected_peak_rate):\n # Pre-create shards\n num_shards = ceil(expected_peak_rate / 1000)\n create_shards(entity_id, counter_type, num_shards)\n\n # Pre-warm caches\n warm_read_cache(entity_id, counter_type)\n\n # Allocate dedicated Redis nodes if needed\n if expected_peak_rate > 50000:\n provision_dedicated_redis(entity_id)\n```\n\n\n---\n\n## 6.4 Idempotency and Deduplication\n\nA critical challenge in distributed systems: what happens when a request is retried?\n\n### The Problem\n\n\n```shell\nUser clicks \"Like\" button\n→ Request sent to server\n→ Network timeout (request actually succeeded)\n→ Client retries\n→ Like counted twice!\n```\n\n\n### Solution 1: Idempotency Keys\n\nClients include a unique key with each request. The server rejects duplicates.\n\n\n```python\ndef increment_idempotent(entity_id, counter_type, idempotency_key, user_id):\n # Check if we've seen this key before\n dedup_key = f\"dedup:{entity_id}:{counter_type}:{idempotency_key}\"\n\n if redis.exists(dedup_key):\n return {\"success\": True, \"duplicate\": True}\n\n # Mark as seen (with TTL to prevent unbounded growth)\n redis.setex(dedup_key, 86400, \"1\") # 24 hour TTL\n\n # Perform the increment\n do_increment(entity_id, counter_type)\n\n return {\"success\": True, \"duplicate\": False}\n```\n\n\n### Solution 2: Actor-Based Deduplication\n\nFor user-initiated counters (likes), track which users have already acted.\n\n\n```python\ndef increment_like(post_id, user_id):\n like_key = f\"likes:{post_id}:users\"\n\n # SADD returns 1 if added, 0 if already existed\n was_added = redis.sadd(like_key, user_id)\n\n if was_added:\n do_increment(post_id, \"likes\")\n return {\"success\": True, \"action\": \"liked\"}\n else:\n return {\"success\": True, \"action\": \"already_liked\"}\n```\n\n\nThis naturally prevents duplicate likes from the same user.\n\n### Solution 3: Event Log Deduplication\n\nIf using event sourcing, deduplicate at the event ingestion layer.\n\n\n```python\ndef ingest_event(event):\n event_id = event[\"idempotency_key\"]\n\n # Kafka/Kinesis handles deduplication within a window\n # For longer windows, check persistent storage\n if event_already_processed(event_id):\n return\n\n # Process event\n apply_event(event)\n mark_event_processed(event_id)\n```\n\n\n### Trade-offs\n\n\n| ##### Approach | ##### Memory | ##### Complexity | ##### Best For |\n| --- | --- | --- | --- |\n| **Idempotency Keys** | High (store all keys) | Low | API-level deduplication |\n| **Actor-Based** | Medium (store user sets) | Low | User actions (likes, votes) |\n| **Event Log** | Low (windowed) | High | Event-sourced systems |\n\n\n---\n\n## 6.5 Read Optimization Strategies\n\nSince reads are typically 10x more frequent than writes, optimizing the read path is critical.\n\n### Strategy 1: Aggressive Caching with Short TTL\n\nCache counter values with a short TTL, accepting slight staleness.\n\n\n```python\ndef read_counter_cached(entity_id, counter_type):\n cache_key = f\"counter:{entity_id}:{counter_type}:total\"\n\n # Try cache first\n cached = cache.get(cache_key)\n if cached is not None:\n return int(cached)\n\n # Cache miss: compute from shards\n total = sum_all_shards(entity_id, counter_type)\n\n # Cache with 5 second TTL\n cache.setex(cache_key, 5, total)\n\n return total\n```\n\n\n**Trade-off:** Values may be up to 5 seconds stale.\n\n### Strategy 2: Background Refresh\n\nProactively refresh popular counters before cache expires.\n\n\n```python\nclass CounterRefresher:\n def __init__(self):\n self.hot_counters = set() # Track frequently accessed counters\n\n def mark_accessed(self, entity_id, counter_type):\n key = f\"{entity_id}:{counter_type}\"\n self.hot_counters.add(key)\n\n def refresh_hot_counters(self):\n \"\"\"Run every second\"\"\"\n for key in self.hot_counters:\n entity_id, counter_type = key.split(\":\")\n total = sum_all_shards(entity_id, counter_type)\n cache.setex(f\"counter:{key}:total\", 10, total)\n\n # Decay: remove counters not accessed recently\n self.hot_counters = get_recently_accessed()\n```\n\n\nThis ensures hot counters are always fresh in cache.\n\n### Strategy 3: Approximate Counts for Display\n\nFor UI display, exact counts often do not matter. Show human-friendly approximations.\n\n\n```python\ndef format_count(value):\n if value < 1000:\n return str(value)\n elif value < 1000000:\n return f\"{value / 1000:.1f}K\" # \"12.5K\"\n elif value < 1000000000:\n return f\"{value / 1000000:.1f}M\" # \"3.2M\"\n else:\n return f\"{value / 1000000000:.1f}B\" # \"1.5B\"\n```\n\n\nWhen displaying \"3.2M likes\", being off by a few thousand does not matter.\n\n### Strategy 4: Batch Reads for Feeds\n\nWhen rendering a feed with many posts, batch all counter reads into one request.\n\n\n```python\ndef get_feed_counters(post_ids):\n \"\"\"Fetch all counters for a feed in one round-trip\"\"\"\n keys = []\n for post_id in post_ids:\n keys.append(f\"counter:{post_id}:likes:total\")\n keys.append(f\"counter:{post_id}:views:total\")\n keys.append(f\"counter:{post_id}:comments:total\")\n\n # Single MGET for all values\n values = cache.mget(keys)\n\n # Parse into structured response\n result = {}\n for i, post_id in enumerate(post_ids):\n result[post_id] = {\n \"likes\": int(values[i * 3] or 0),\n \"views\": int(values[i * 3 + 1] or 0),\n \"comments\": int(values[i * 3 + 2] or 0),\n }\n\n return result\n```\n\n\nThis fetches 100 counters in 1 Redis round-trip instead of 100.\n\n---\n\n## 6.6 Failure Handling and Recovery\n\nWhat happens when parts of the system fail?\n\n### Redis Failure\n\n**Scenario:** Redis cluster becomes unavailable\n\n**Impact:** Cannot write or read counter values\n\n**Mitigation:**\n\n1. **Fallback to database:** Serve reads from persistent storage (slower but available)\n2. **Queue writes:** Buffer increments in a message queue (Kafka) until Redis recovers\n3. **Circuit breaker:** Fail fast and return cached/approximate values\n\n\n```python\ndef increment_with_fallback(entity_id, counter_type):\n try:\n return redis_increment(entity_id, counter_type)\n except RedisUnavailable:\n # Queue for later processing\n kafka.produce(\"counter_increments\", {\n \"entity_id\": entity_id,\n \"counter_type\": counter_type,\n \"delta\": 1,\n \"timestamp\": time.time()\n })\n return {\"success\": True, \"queued\": True}\n```\n\n\n### Write-Ahead Log Failure\n\n**Scenario:** Kafka becomes unavailable\n\n**Impact:** Cannot durably record writes\n\n**Mitigation:**\n\n1. **Synchronous database writes:** Fall back to direct database updates (slower)\n2. **Local disk buffer:** Write to local disk, replay when Kafka recovers\n3. **Reject writes:** Return 503, let clients retry\n\n### Database Failure\n\n**Scenario:** Primary database is down\n\n**Impact:** Cannot persist counter values, cannot recover from cold start\n\n**Mitigation:**\n\n1. **Read replicas:** Promote a replica to primary\n2. **Continue in-memory:** Redis has recent values, persist when DB recovers\n3. **WAL replay:** Replay from Kafka when database is back\n\n### Recovery Process\n\nAfter a failure, the system must reconcile state:\n\n\n```python\ndef recover_counter(entity_id, counter_type):\n # 1. Load last known value from database\n db_value = database.get_counter(entity_id, counter_type)\n db_version = database.get_version(entity_id, counter_type)\n\n # 2. Replay WAL entries after that version\n wal_entries = kafka.consume_from(\n topic=\"counter_increments\",\n filter=lambda e: e[\"entity_id\"] == entity_id and e[\"counter_type\"] == counter_type,\n after_version=db_version\n )\n\n delta = sum(e[\"delta\"] for e in wal_entries)\n\n # 3. Initialize Redis with recovered value\n recovered_value = db_value + delta\n redis.set(f\"counter:{entity_id}:{counter_type}:shard:0\", recovered_value)\n\n # 4. Update read cache\n read_cache.set(f\"counter:{entity_id}:{counter_type}:total\", recovered_value)\n```\n\n\n---\n\n## References\n\n- [Building Real-Time Analytics at Scale](undefined) - Facebook Engineering on high-throughput counting systems\n- [Conflict-free Replicated Data Types](undefined) - Academic resource on CRDTs including G-Counters and PN-Counters\n- [Sharded Counters in Google Cloud](undefined) - Google's documentation on distributed counters in Firestore\n- [How We Scaled Our Counter Infrastructure](undefined) - LinkedIn Engineering on scaling counters to billions of events\n- [Redis Best Practices for Counters](undefined) - Redis documentation on atomic counter operations\n\n---\n\n# Quiz","pageType":"system-design-interviews"}

Design Distributed Counter

Ashish Pratap Singh

Get Premium