Requirements

Functional Requirements:

List the key functional requirements for the system (Ask the AI for hints if stuck)...

• The counter can decrement if requested
• The counter can increment if requested
• The decrement and increment can be requested by clients distributed across regions
• The user can request to see updated counts in real-time

Non-Functional Requirements:

List the key non-functional requirements (performance, scalability, reliability, etc.)...

• The system should be available 99.99% of time
• The p99 latency for incrementing and decrementing count should be less than 200ms, and the p99 latency for viewing updated count should be less than 100ms
• The system can be eventually consistent, while not everyone see the updated counts simultaneously.
• The system should be easily scalable to more users and regions.

Capacity Estimation

Estimate the scale of the system. Consider daily active users, read/write ratio, storage requirements, bandwidth, and any relevant QPS calculations...

Assuming that we have 1 million new events across several regions each second, and 100 million people are refreshing the counts simultaneously in peak hours.

Peak write QPS = 1 million

Peak read QPS = 100 million

For storage, we don't need to store all the increment/decrement events, so we can later do an audit review, as well as the distributed counters. The major storage is all the events.

Assuming each event takes 1kb to store, and on average throughout the day, each second takes in 100 thousand events.

Storage needed per day = 100 k events per second * 100k seconds per day * 1kb = 10 billion * 1kb = 10TB of additional storage per day. For counters that have expired and stopped, we put the events storage to cold storage like S3.

For network bandwidth, in peak hours, peak read QPS = 100 million. Assuming that we transfer 1kb per request per second, peak network bandwidth is roughly:

1kb * 100 million = 100Gb per second

API Design

Define the APIs expected from the system. This is your chance to analyze and define the read and write paths so that you can come up with the high-level design...

The APIs for this system is relatively simple, CRUD on the counts:

GET v1/view_counts

POST v1/increment

POST v1/decrement

High-Level Design

Describe the overall system architecture. Identify the main components needed to solve the problem end-to-end. Use the diagramming tool to create a block diagram.

In this design, the write path is the path to update counters.

We take in events from the event ingestion source, and the global load balancers redirect traffic to API gateways in different global regions.

For the API gateways, they do:

1. Authorization and authentication
2. Rate limiting
3. Idempotent check, to avoid duplicate events

Then the requests are forwarded to the counter services in each region. The counter service determines the shard for each event for each video, breaking up the count, and publish event on regional Kafkas. Event is like:

event_id, video_id, shard_id, timestamp, increment, decrement

The counters workers servers in different regions process the events. And do 2 things:

1. They write the current sharded counts to redis clusters in each regions. Every 5 seconds, a periodic checkpoint flushes the latest sharded counts and the kafka offsets to cassandra. In case of redis cluster failure, we can use the kafka offset + existing checkpoint storage in cassandra to recover the data.
2. They publish events to another topic topic. The events contain the current sharded count values of each video in different regions. A kafka cluster publishes these events, and they get processed by downstreams regional aggregators. The aggregators then aggregate the counts and write them to redis cluster again. This is the final global count used in read path.

On the read path, after load balancers and API gateway, users read updated global counts from their regional redis cluster.

For the redis cluster, both the regional sharded count and global counts are cached. For hot videos, we use below strategies:

1. We still shard the global counts, and aggregate in read time, preventing traffic from overwhelming a single redis shard.
2. We use TTL jittering, request coalescing and stale-while-revalidate to prevent thundering herd when popular keys expire.

Database Design

Define the data model. Identify the main entities, their attributes, and relationships. Consider the choice of database type (SQL vs NoSQL) and justify your decision based on access patterns...

In this system, there are three types of data to store:

1. Raw increment/decrement events
2. Regional counter shard state and recovery checkpoints
3. Materialized global counts for fast reads

1. Raw count events — Kafka

Each increment/decrement request is first written to Kafka.

Example event:

Kafka acts as the durable event log.

We partition events using:

This guarantees that all events for the same physical counter shard are processed in order by a single consumer within a consumer group.

For very popular videos, we divide the logical counter into multiple shards:

This prevents a single Kafka partition or Redis key from becoming a bottleneck.

2. Regional live counter state — Redis

Shard workers consume events from Kafka and apply them to regional Redis using atomic INCR / DECR operations.

The key can be modeled as:

For example:

Redis is used because:

1. Increment/decrement operations are atomic and extremely low latency.
2. It supports very high throughput.
3. It can be horizontally partitioned across multiple Redis shards.
4. The latest count becomes visible immediately after the worker applies the event.

Redis is not the only durable source of truth. Kafka retains the event history required to reconstruct state after failures.

3. Durable checkpoints — Cassandra

Periodically, for example every 5 seconds, the current Redis shard state is checkpointed into Cassandra.

The Cassandra table can be modeled as:

The partition key is:

This distributes shards of a popular video across Cassandra nodes and avoids creating a hot Cassandra partition.

Cassandra is suitable because:

1. Checkpoints must be durable so that Redis can be recovered after failure.
2. It supports high write throughput and horizontal scalability.
3. The workload is simple key-based reads and writes.
4. Eventual consistency is acceptable for checkpoints because Kafka remains the durable event log.

We use a replication factor of 3 across availability zones.

For example:

A checkpoint write must therefore be acknowledged by 2 of the 3 replicas.

Reads may return a slightly stale checkpoint, which is acceptable because recovery can replay Kafka events after the stored offset.

Recovery works as:

4. Global materialized count — Redis

Each region has an aggregator that receives regional counter updates through the aggregation stream.

The aggregator combines regional shard counts into a global count:

The result is written to a regional Redis cluster:

For example:

This Redis cluster directly serves read traffic.

Reads are therefore:

This provides very low read latency.

Hot counters

For extremely popular videos, the logical counter is sharded at the write layer.

For example:

Each shard can be processed independently by Kafka workers and stored on different Redis nodes.

The aggregator periodically combines these shard values into one materialized global count.

The global Redis value itself is a single key, but because reads are cheap and can be replicated, this is usually less problematic than a hot write key. If one key becomes exceptionally read-heavy, Redis replicas or application-level read replication can absorb the load.

Detailed Component Design

Deep dive into 2-3 key components. Explain how they work, how they scale, discuss tradeoffs, capacity, and any relevant algorithms or data structures.

In this system, there are three types of data to store:

1. Raw increment/decrement events
2. Regional counter shard state and recovery checkpoints
3. Materialized global counts for fast reads

1. Raw count events — Kafka

Each increment/decrement request is first written to Kafka.

Example event:

Kafka acts as the durable event log.

We partition events using:

This guarantees that all events for the same physical counter shard are processed in order by a single consumer within a consumer group.

For very popular videos, we divide the logical counter into multiple shards:

This prevents a single Kafka partition or Redis key from becoming a bottleneck.

2. Regional live counter state — Redis

Shard workers consume events from Kafka and apply them to regional Redis using atomic INCR / DECR operations.

The key can be modeled as:

For example:

Redis is used because:

1. Increment/decrement operations are atomic and extremely low latency.
2. It supports very high throughput.
3. It can be horizontally partitioned across multiple Redis shards.
4. The latest count becomes visible immediately after the worker applies the event.

Redis is not the only durable source of truth. Kafka retains the event history required to reconstruct state after failures.

3. Durable checkpoints — Cassandra

Periodically, for example every 5 seconds, the current Redis shard state is checkpointed into Cassandra.

The Cassandra table can be modeled as:

The partition key is:

This distributes shards of a popular video across Cassandra nodes and avoids creating a hot Cassandra partition.

Cassandra is suitable because:

1. Checkpoints must be durable so that Redis can be recovered after failure.
2. It supports high write throughput and horizontal scalability.
3. The workload is simple key-based reads and writes.
4. Eventual consistency is acceptable for checkpoints because Kafka remains the durable event log.

We use a replication factor of 3 across availability zones.

For example:

A checkpoint write must therefore be acknowledged by 2 of the 3 replicas.

Reads may return a slightly stale checkpoint, which is acceptable because recovery can replay Kafka events after the stored offset.

Recovery works as:

4. Global materialized count — Redis

Each region has an aggregator that receives regional counter updates through the aggregation stream.

The aggregator combines regional shard counts into a global count:

The result is written to a regional Redis cluster:

For example:

This Redis cluster directly serves read traffic.

Reads are therefore:

This provides very low read latency.

Hot counters

For extremely popular videos, the logical counter is sharded at the write layer.

For example:

Each shard can be processed independently by Kafka workers and stored on different Redis nodes.

The aggregator periodically combines these shard values into one materialized global count.

The global Redis value itself is a single key, but because reads are cheap and can be replicated, this is usually less problematic than a hot write key. If one key becomes exceptionally read-heavy, Redis replicas or application-level read replication can absorb the load.

In this design, the write path is the path to update counters.

We take in events from the event ingestion source, and the global load balancers redirect traffic to API gateways in different global regions.

For the API gateways, they do:

1. Authorization and authentication
2. Rate limiting
3. Idempotent check, to avoid duplicate events

Then the requests are forwarded to the counter services in each region. The counter service determines the shard for each event for each video, breaking up the count, and publish event on regional Kafkas. Event is like:

event_id, video_id, shard_id, timestamp, increment, decrement

The counters workers servers in different regions process the events. And do 2 things:

1. They write the current sharded counts to redis clusters in each regions. Every 5 seconds, a periodic checkpoint flushes the latest sharded counts and the kafka offsets to cassandra. In case of redis cluster failure, we can use the kafka offset + existing checkpoint storage in cassandra to recover the data.
2. They publish events to another topic topic. The events contain the current sharded count values of each video in different regions. A kafka cluster publishes these events, and they get processed by downstreams regional aggregators. The aggregators then aggregate the counts and write them to redis cluster again. This is the final global count used in read path.

On the read path, after load balancers and API gateway, users read updated global counts from their regional redis cluster.

For the redis cluster, both the regional sharded count and global counts are cached. For hot videos, we use below strategies:

1. We still shard the global counts, and aggregate in read time, preventing traffic from overwhelming a single redis shard.
2. We use TTL jittering, request coalescing and stale-while-revalidate to prevent thundering herd when popular keys expire.

For each increment/decrement event, we choose one of the video's counter shards. We don't permanently map a video_id to one shard—that would defeat the purpose of sharding a hot counter.

Suppose video_123 has 64 shards:

A clean strategy is:

For example:

Then the Counter Service publishes:

and uses this as the Kafka key:

So the flow is: