System requirements

Functional:

The counter will be triggered by user actions on a massive scale social network. Use youtube as example.

Clients should be able to increment a value globally.

Clients should be able to retrieve the global value.

Non-Functional:

Counter value is realtime.

All operations should be fast, even on highly concurrent write scenario.

Counter value should be accurate.

Eventual consistency.

Highly available, scalable.

Capacity estimation

Assumption:

• the system should be able to handle a hot video with 20 million views per hour
• 1 billion daily active users
• 1 user views 10 videos a day
• The webpage will fetch view count 10 times during the time a user views a video.
• 20 byte per counter in storage
• we need to store the counter for 5 years.

Peak Write QPS per counter = 20M/h = 5k

Average Write QPS 1B user * 10 video views / 24h = 120k

Average Read QPS = 10 read per write = 1M

Storage = 1B * 1 Video per day * 5 year * 20 byte = 36Tb

API design

GET /v1/counter/

parameters:counter_id string

return: value integer

POST /v1/counter/

parameters:

object_id: string - the id of the object that this counter is associated to

activity_type: string - the type of activity that this counter captures

return: counter_id string

PUT /v1/counter/

parameters:

counter_id: string

return:

none

Database design

Counter DB (No SQL)

This component is responsible to store all the counters and values.

The data has no relation.

I would go with no sql, consist of

Counter table

counter_id (string),

counter_value (integer )

I would prefer a nosql db with key-value store like access interface(Cassandra for example), over in-memory db like redis, as not every counter will be read frequently, hence using in-memory only storage would be a waste of resource. DB with persistence might be slower than Redis, but is more cost-effective.

We can use Redis to cache hot counter data to balance latency vs cost.

Hot Counter DB (noSQL)

(this db stores hot counters that will be split into multiple shards).

Hot Counter table

counter_id : string

num_shards: integer

This data will store the hot counter ids, used by counter service to know know if a counter id is hot counter, and will need to be routed to multiple shards. The data will be small, and read should be blazingly fast. I prefer redis here.

High-level design

Stateless API service that takes get/post requests from user.

Request hitting Counter service will be protected via rate limiter in API gateway.

API servers are stateless and globally available.

Hot Counter DB stores information used by Counter Service, to direct write traffic of hot counter to one of the assigned shards; and aggregate results from all shards from all shards for read requests.

Sharded Counter DB stores values for all counters.

Redis as look aside cache for API servers cache counter values from db.

Async traffic monitoring tasks monitors updates to CounterDB, detect unexpected hot counter, and add it to hot counter db, so unexpected hot counter won't experience long write contention.

Request flows

Write:

1. counter service checks the counter id with hot counter db
2. if counter is in hot counter db, append the counter_id with a number in range [0, num_shard]
3. increment counter id in CounterDB

Read:

1. counter service checks if counter exists in cache and return the value from cache if it exists
2. otherwise, counter service fetch the counter value from CounterDB
   1. if counter is not a hot counter, fetch the value with counter_id from counterDb
   2. if counter is a hot counter, fetch number of shards from hot counter db
   3. for num in [0, num_shard], fetch value from CounterDb with counter_id concatenated with num.
   4. sum up the value from all concatenated counters
3. store value in cache and return value as response

Detailed component design

1.Counter Service

Upon write, Counter service calls finds the right DB host for the counter to increment the value, based on the above information.

Upon read, if the value is not present in cache, the api server collect value from all shards and sum them up and store the value in Redis Cache.

2.DB Design

Ideally the db will use leaderless or multi-leader replication to ensure high availability, as well as CRDT to ensure data consistency.

2.1. Handling Write Request

2.1.1 To ensure normal write requests are fast, each data center has at least a leader per shard. Write requests are routed to the nearest data center. Synchronization between leader and followers within the same data center is fast. Data is asynchronously replicated between data centers.

This will raise issue in data consistency that I'll cover later.

2.1.2 For hot counter, a routing service will route the request to an available shard, hence the write load is split across multiple nodes, and write contention is reduced.

2.2.Handling Read

Read request could be served from replica to alleviate load from leaders.

2.3.Handling Data Consistency

DB with Last write win like Cassandra or DynamoDB isn't a good choice as this will cause significant data loss when leaders concurrently receives lots of writes during write spike, and we want data to be accurate.

The db should adopt conflict free replicated data type(CRDT) to sync up between leaders. This is particularly suitable for counter operations, as number of increment and decrement will converge eventually across leaders, regardless of sequence.

3. Redis Cache Design

Cache Invalidation

Invalidate cache upon every write request is bad idea, which will cause write contention, especially upon allowed traffic burst. Instead, we'll have do the invalidation asynchronously.

We can have a worker set that processes DB changes via techniques like CDC, and remove the counter that was recently changed from cache.

Trade offs/Tech choices

1. I chose a synchronous write path, over asynchronous processing architecture, so that users can get the counter value real time.
2. I choose to make read path parallel, hence can be potentially slower, in order to be able to split write traffic across multiple shards. Writing is more contentious than read.

Failure scenarios/bottlenecks

1. Common failure scenarios like leader failure, follower failure, data center outage etc, are handled by individual storage system techniques.
2. Write Contention: hot counters will experience high concurrent writes. The design address this by splitting the write traffic into multiple shards.
3. Double counting: if a write to db succeeds but fails to respond to client, client will retry the write hence cause double counting.
4. Generally applications should have rate limiter either in API gateway or at API service, to ensure system availability upon traffic spike.

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?

Potentially have a request table, to record the latest committed write request. This ensure that a single request doesn't get commited twice. This will help implement idempotency in write path, to avoid double counting from partial failure.