Requirements

Functional Requirements:

1. Create distributed counters and keep the increment/decrement counts.
2. Store the delta, exact activity feed is not needed
3. Display the counter to the users (Assuming 1.5x read: write ratio on an average)

Non-Functional Requirements:

1. High Availability: The system should be highly available (AP over CP during partition)
2. High Scalability: Handle DAU of 100M users. Assuming 10% concurrency, we can have 10M users activity at a time.
3. The counters can be eventually consistent
4. Highly Durable: Once the activity is recorded, it should not be lost

Capacity Estimation

Active DAU = 100M

Concurrent user = 10% = 10M

Assuming we have an activity recorded every 5 sec / user

Writes/sec = 10*10^6/5= 2M writes/sec

Peak writes = 6M writes/sec

Reads/sec = 3M/sec

Peak reads = 9M reads/sec

Storage: The counter will be 1 counter / entity.

The counter + metadata could be roughly 250bytes.

Total storage for a given day = 100M * 250 bytes = 25GB (however, this is not the right number to scale, for estimating monthly and yearly storage, as it will be dependent of repeat user). We expect this number to grow relatively slowly.

The storage requirement is trivial

API Design

POST /v1/increment

POST /v1/decrement

[Both have same body]

Headers:

Authentication: Bearer

Content: "application/json"

Accept: "application/json"

Idempotency-Key: uuid

Body: {

user_id: uuid,

content_id: uuid

}

Response:

Headers:

RateLimit-Limit: 100

RateLimit-RetryAfter: 30

{

user_id: uuid,

content_id: uuid,

curr_count: int

}

Status: 200 (Success), 429 (Rate Limited), 500 (Server Error)

GET /v1/counter/{content_id}

Authentication: Bearer

Content: "application/json"

Accept: "application/json"

Idempotency-Key: uuid

If-Not-Match: uuid

Response:

{

curr_count: int

}

Status: 200 (Success), 429 (Rate Limited), 500 (Server Error)

High-Level Design

The clients request is routed through Load Balancer and Gateway (we also check the Rate Balancing here)

Write:

Write Services dumps the request into a Kafka and return 202 Accepted status to the user.

An Aggregator service reads the Kafka topics, aggregates the count and updates the DB.

Kafka topic is partitioned by content_id, so the activities of a particular content is consumed parallel

Read:

Read is primarily served from Redis and then fallsback to Cassandra as a source of truth (Cache aside method)

We can use LRU as a cache invalidation method

Hot Key Handling:

For hot keys, we can partition the kafka topics using content_id + salt. Use locks to upgrade the Cassandra.

Database Design

Cassandra:

{

content_id: uuid, [PK]

counter: int,

last_updated_at: timestamp

}

Partition Key: (content_id, timestamp)

Redis:

Key: counter:{content_id}

Value: int

Aggregator Table:

{

content_id: uuid,

shard_id: uuid,

bucket: timestamp,

batch: uuid,

delta: int

}

Detailed Component Design

1. To handle the high write throughput, we seggregate the Write service acknowledgement with DB persistence. Instead of making a DB call for each event, we aggregate the events and batch update Cassandra. We chose cassandra, bcz our data is a simple key-value type lookup and has a high write throughput.
2. Since Kafka is durable and has logs to retry, we can have background services, that continuously replays the events and flags any deviation in the overall count in the Cassandra. For any failure during the aggregation, the event can be retried and ultimately end up in a DLQ, which can be later handled by the Ops team.
3. To keep the Reads fast, we cache the result in Redis Cluster. Also, we can cache the data in CDN, with a TTL of 60 sec. Since we can afford a little latency, this should not be an issue.

Failure Scenarios:

1. Kafka Goes Down: If the kafka partition goes down, we still have the data replicated.
2. Aggregator service goes down: Kafka has durable logs, which can be used to replay the events and catch up.
3. Redis cluster going down: Cassandra is the fallback. We also do have cache in CDN, which can be used to serve the data. We can limit the number of requests going to Cassandra if the Redis Cluster failure persists, and use switch breakers, in order to prevent the failures from cascading

Handling Hot Keys:

To handle hot keys, we will partition the Kafka by content_id + salt, this way, events can be processed parallelly. In the aggregation table, we can store the current aggregate as content_id:{shard_1}... content_id:{shard_n} -> aggregate and then flush the final data to cassandra using locks on the content id